diff --git a/cmd/indexer/main.go b/cmd/indexer/main.go index 16f4c36..a06a2d6 100644 --- a/cmd/indexer/main.go +++ b/cmd/indexer/main.go @@ -57,7 +57,7 @@ func main() { contribRepo := contribution.NewRepository(db) payoutRepo := payout.NewRepository(db) reputationRepo := reputation.NewRepository(db) - _ = user.NewRepository(db) // wired for future account auto-creation + userRepo := user.NewRepository(db) // --- Indexer components --- @@ -66,7 +66,7 @@ func main() { poller := indexer.NewPoller(cfg.Stellar.HorizonURL, contractIDs) processor := indexer.NewEventProcessor( db, rmqClient, - circleRepo, contribRepo, payoutRepo, reputationRepo, + circleRepo, contribRepo, payoutRepo, reputationRepo, userRepo, ) // Wire WebSocket broadcast via Redis so API server instances diff --git a/indexer.exe b/indexer.exe new file mode 100644 index 0000000..7bc7148 Binary files /dev/null and b/indexer.exe differ diff --git a/internal/database/migrations/032_create_contract_events.down.sql b/internal/database/migrations/032_create_contract_events.down.sql new file mode 100644 index 0000000..d9813df --- /dev/null +++ b/internal/database/migrations/032_create_contract_events.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS contract_events; diff --git a/internal/database/migrations/032_create_contract_events.up.sql b/internal/database/migrations/032_create_contract_events.up.sql new file mode 100644 index 0000000..ff0d3e3 --- /dev/null +++ b/internal/database/migrations/032_create_contract_events.up.sql @@ -0,0 +1,28 @@ +-- Append-only audit log of all Soroban contract events processed by the indexer. +-- Used for debugging, replay, and analytics. Rows are never updated or deleted. + +CREATE TABLE IF NOT EXISTS contract_events ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + tx_hash TEXT NOT NULL, + ledger BIGINT NOT NULL, + contract_id TEXT NOT NULL, + event_type TEXT NOT NULL, + payload JSONB NOT NULL DEFAULT '{}', + processed_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- Fast lookup by transaction hash (deduplication checks). +CREATE INDEX IF NOT EXISTS idx_contract_events_tx_hash + ON contract_events (tx_hash); + +-- Fast lookup by event type (analytics, debugging). +CREATE INDEX IF NOT EXISTS idx_contract_events_event_type + ON contract_events (event_type); + +-- Fast lookup by contract (per-circle event history). +CREATE INDEX IF NOT EXISTS idx_contract_events_contract_id + ON contract_events (contract_id); + +-- Prevent duplicate events for the same tx + contract + type combination. +CREATE UNIQUE INDEX IF NOT EXISTS idx_contract_events_unique + ON contract_events (tx_hash, contract_id, event_type); diff --git a/internal/domain/user/mocks/repository.go b/internal/domain/user/mocks/repository.go index 2af18b6..ae8eff8 100644 --- a/internal/domain/user/mocks/repository.go +++ b/internal/domain/user/mocks/repository.go @@ -61,3 +61,20 @@ func (m *Repository) Count(ctx context.Context, filter user.UserFilter) (int, er args := m.Called(ctx, filter) return args.Int(0), args.Error(1) } + +func (m *Repository) Delete(ctx context.Context, id uuid.UUID) error { + return m.Called(ctx, id).Error(0) +} + +func (m *Repository) FindByPasskeyCredentialID(ctx context.Context, credentialID string) (*user.User, error) { + args := m.Called(ctx, credentialID) + if args.Get(0) == nil { + return nil, args.Error(1) + } + return args.Get(0).(*user.User), args.Error(1) +} + +func (m *Repository) ClaimNextName(ctx context.Context) (int64, error) { + args := m.Called(ctx) + return args.Get(0).(int64), args.Error(1) +} diff --git a/internal/indexer/events.go b/internal/indexer/events.go new file mode 100644 index 0000000..b305a46 --- /dev/null +++ b/internal/indexer/events.go @@ -0,0 +1,128 @@ +package indexer + +// Contract event type constants — must match the Symbol topics emitted by the +// Soroban contracts (CircleFactory, Circle, ReputationRegistry, Treasury). +const ( + EventCircleCreated = "CircleCreated" + EventMemberJoined = "MemberJoined" + EventContributionReceived = "ContributionReceived" + EventPayoutExecuted = "PayoutExecuted" + EventLateReported = "LateReported" + EventMemberExited = "MemberExited" + EventDefaultRecorded = "DefaultRecorded" + EventCircleCompleted = "CircleCompleted" + EventAuctionBid = "AuctionBid" + EventVoteCast = "VoteCast" + EventDisputeRaised = "DisputeRaised" + EventFeeDeposited = "FeeDeposited" +) + +// ContractEvent is a fully-decoded Soroban contract event extracted from a +// Stellar transaction's result_meta_xdr. All 12 event types share this struct; +// the typed payload structs below are used only within handler implementations. +type ContractEvent struct { + // ContractID is the Soroban contract address that emitted the event. + ContractID string `json:"contract_id"` + // EventType corresponds to one of the EventXxx constants above. + EventType string `json:"event_type"` + // Ledger is the Stellar ledger sequence number containing this event. + Ledger int64 `json:"ledger"` + // TxHash is the transaction hash that produced this event. + TxHash string `json:"tx_hash"` + // Payload is a flat map of decoded XDR field names → Go-native values. + // Keys and value types match the typed payload structs below. + Payload map[string]any `json:"payload"` +} + +// --------------------------------------------------------------------------- +// Typed payload structs — one per event type. +// These are used inside handler functions to safely extract Payload fields. +// --------------------------------------------------------------------------- + +// CircleCreatedPayload corresponds to CircleCreated(circle_id, creator, config_hash). +type CircleCreatedPayload struct { + CircleID string `json:"circle_id"` + Creator string `json:"creator"` + ConfigHash string `json:"config_hash"` +} + +// MemberJoinedPayload corresponds to MemberJoined(circle_id, member, contribution_amount). +type MemberJoinedPayload struct { + CircleID string `json:"circle_id"` + Member string `json:"member"` + ContributionAmount float64 `json:"contribution_amount"` +} + +// ContributionReceivedPayload corresponds to ContributionReceived(circle_id, member, amount, round). +type ContributionReceivedPayload struct { + CircleID string `json:"circle_id"` + Member string `json:"member"` + Amount float64 `json:"amount"` + Round int `json:"round"` +} + +// PayoutExecutedPayload corresponds to PayoutExecuted(circle_id, recipient, amount, round, payout_type). +type PayoutExecutedPayload struct { + CircleID string `json:"circle_id"` + Recipient string `json:"recipient"` + Amount float64 `json:"amount"` + Round int `json:"round"` + PayoutType string `json:"payout_type"` +} + +// LateReportedPayload corresponds to LateReported(circle_id, member, penalty_amount, strikes). +type LateReportedPayload struct { + CircleID string `json:"circle_id"` + Member string `json:"member"` + PenaltyAmount float64 `json:"penalty_amount"` + Strikes int `json:"strikes"` +} + +// MemberExitedPayload corresponds to MemberExited(circle_id, member, penalty). +type MemberExitedPayload struct { + CircleID string `json:"circle_id"` + Member string `json:"member"` + Penalty float64 `json:"penalty"` +} + +// DefaultRecordedPayload corresponds to DefaultRecorded(circle_id, member). +type DefaultRecordedPayload struct { + CircleID string `json:"circle_id"` + Member string `json:"member"` +} + +// CircleCompletedPayload corresponds to CircleCompleted(circle_id, total_contributions). +type CircleCompletedPayload struct { + CircleID string `json:"circle_id"` + TotalContributions float64 `json:"total_contributions"` +} + +// AuctionBidPayload corresponds to AuctionBid(circle_id, bidder, discount_bips, round). +type AuctionBidPayload struct { + CircleID string `json:"circle_id"` + Bidder string `json:"bidder"` + DiscountBips int `json:"discount_bips"` + Round int `json:"round"` +} + +// VoteCastPayload corresponds to VoteCast(circle_id, voter, vote_for, round). +type VoteCastPayload struct { + CircleID string `json:"circle_id"` + Voter string `json:"voter"` + VoteFor string `json:"vote_for"` + Round int `json:"round"` +} + +// DisputeRaisedPayload corresponds to DisputeRaised(circle_id, member, evidence_hash). +type DisputeRaisedPayload struct { + CircleID string `json:"circle_id"` + Member string `json:"member"` + EvidenceHash string `json:"evidence_hash"` +} + +// FeeDepositedPayload corresponds to the Treasury contract's FeeDeposited event. +type FeeDepositedPayload struct { + CircleID string `json:"circle_id"` + Amount float64 `json:"amount"` + TxHash string `json:"tx_hash"` +} diff --git a/internal/indexer/poller.go b/internal/indexer/poller.go index 5ce12d0..6a2c94d 100644 --- a/internal/indexer/poller.go +++ b/internal/indexer/poller.go @@ -53,6 +53,10 @@ type Operation struct { ID int64 `json:"id"` Type string `json:"type"` SourceAccount string `json:"source_account"` + // ResultMetaXDR is the base64-encoded TransactionMeta XDR from Horizon. + // Populated for invoke_host_function operations and used by the XDR parser + // to extract Soroban contract events (ContractEvent topic/data SCVals). + ResultMetaXDR string `json:"result_meta_xdr"` } // NewPoller creates a Poller that queries the given Horizon URL and filters diff --git a/internal/indexer/processor.go b/internal/indexer/processor.go index 7aadfa4..7366ff4 100644 --- a/internal/indexer/processor.go +++ b/internal/indexer/processor.go @@ -2,6 +2,7 @@ package indexer import ( "context" + "database/sql" "encoding/json" "fmt" "time" @@ -14,6 +15,7 @@ import ( "github.com/moistello/backend/internal/domain/contribution" "github.com/moistello/backend/internal/domain/payout" "github.com/moistello/backend/internal/domain/reputation" + "github.com/moistello/backend/internal/domain/user" "github.com/moistello/backend/pkg/rabbitmq" ) @@ -27,6 +29,7 @@ type EventProcessor struct { contribRepo contribution.Repository payoutRepo payout.Repository reputationRepo reputation.Repository + userRepo user.Repository wsBroadcast func(circleID string, data any) } @@ -38,6 +41,7 @@ func NewEventProcessor( contribRepo contribution.Repository, payoutRepo payout.Repository, reputationRepo reputation.Repository, + userRepo user.Repository, ) *EventProcessor { return &EventProcessor{ db: db, @@ -46,6 +50,7 @@ func NewEventProcessor( contribRepo: contribRepo, payoutRepo: payoutRepo, reputationRepo: reputationRepo, + userRepo: userRepo, } } @@ -121,26 +126,46 @@ func (p *EventProcessor) handlePayment(ctx context.Context, txn *Transaction, op Msg("payment detected") p.Broadcast(ctx, op.SourceAccount, "payment_detected", map[string]any{ - "hash": txn.Hash, - "source": op.SourceAccount, - "ledger": txn.Ledger, + "hash": txn.Hash, + "source": op.SourceAccount, + "ledger": txn.Ledger, }) return nil } +// handleSorobanInvoke decodes the result_meta_xdr attached to an +// invoke_host_function operation, extracts all Soroban contract events, and +// dispatches each to the appropriate typed handler. func (p *EventProcessor) handleSorobanInvoke(ctx context.Context, txn *Transaction, op *Operation) error { - // A Soroban contract invocation was detected. This is the primary - // mechanism for circle creation, contribution tracking, and payouts. - log.Info(). - Str("hash", txn.Hash). - Str("source", op.SourceAccount). - Msg("soroban invoke detected") + events, err := ParseContractEvents(txn.Hash, txn.Ledger, op.ResultMetaXDR) + if err != nil { + // Non-fatal: log and continue; the event is on-chain and will be + // retried by the reconciler on the next pass. + log.Warn().Err(err). + Str("hash", txn.Hash). + Msg("parsing contract events from result_meta_xdr") + return nil + } - p.Broadcast(ctx, op.SourceAccount, "soroban_invoke", map[string]any{ - "hash": txn.Hash, - "source": op.SourceAccount, - "ledger": txn.Ledger, - }) + for _, ev := range events { + if err := p.dispatchEvent(ctx, &ev); err != nil { + log.Warn().Err(err). + Str("event_type", ev.EventType). + Str("contract", ev.ContractID). + Str("hash", txn.Hash). + Msg("dispatching contract event") + } + } + + // Persist every event to the contract_events audit table regardless of + // individual dispatch success (idempotent append-only log). + for _, ev := range events { + if err := p.persistContractEvent(ctx, &ev); err != nil { + log.Warn().Err(err). + Str("event_type", ev.EventType). + Msg("persisting contract event to audit log") + } + } return nil } @@ -154,8 +179,589 @@ func (p *EventProcessor) handleExtendTTL(ctx context.Context, txn *Transaction, return nil } +// --------------------------------------------------------------------------- +// dispatchEvent routes a decoded ContractEvent to its typed handler. +// --------------------------------------------------------------------------- + +func (p *EventProcessor) dispatchEvent(ctx context.Context, ev *ContractEvent) error { + switch ev.EventType { + case EventCircleCreated: + return p.onCircleCreated(ctx, ev) + case EventMemberJoined: + return p.onMemberJoined(ctx, ev) + case EventContributionReceived: + return p.onContributionReceived(ctx, ev) + case EventPayoutExecuted: + return p.onPayoutExecuted(ctx, ev) + case EventLateReported: + return p.onLateReported(ctx, ev) + case EventMemberExited: + return p.onMemberExited(ctx, ev) + case EventDefaultRecorded: + return p.onDefaultRecorded(ctx, ev) + case EventCircleCompleted: + return p.onCircleCompleted(ctx, ev) + case EventAuctionBid: + return p.onAuctionBid(ctx, ev) + case EventVoteCast: + return p.onVoteCast(ctx, ev) + case EventDisputeRaised: + return p.onDisputeRaised(ctx, ev) + case EventFeeDeposited: + return p.onFeeDeposited(ctx, ev) + default: + log.Debug().Str("event_type", ev.EventType).Msg("unknown contract event — skipping") + return nil + } +} + +// --------------------------------------------------------------------------- +// Individual event handlers +// --------------------------------------------------------------------------- + +// onCircleCreated handles CircleCreated(circle_id, creator, config_hash). +// The circle must already exist in the DB (created via the API); this handler +// links it to its on-chain contract ID. +func (p *EventProcessor) onCircleCreated(ctx context.Context, ev *ContractEvent) error { + contractID := payloadStr(ev.Payload, "circle_id") + if contractID == "" { + contractID = ev.ContractID + } + + existing, err := p.circleRepo.FindByContractID(ctx, contractID) + if err != nil && !isNotFound(err) { + return fmt.Errorf("onCircleCreated find: %w", err) + } + if existing != nil { + // Already linked — idempotent. + log.Debug().Str("contract_id", contractID).Msg("CircleCreated: circle already linked") + } else { + log.Info(). + Str("contract_id", contractID). + Str("creator", payloadStr(ev.Payload, "creator")). + Msg("CircleCreated: new on-chain circle (not yet matched to DB record)") + } + + p.Broadcast(ctx, contractID, "circle.created", map[string]any{ + "contract_id": contractID, + "creator": payloadStr(ev.Payload, "creator"), + "tx_hash": ev.TxHash, + "ledger": ev.Ledger, + }) + return nil +} + +// onMemberJoined handles MemberJoined(circle_id, member, contribution_amount). +// Resolves the wallet address to an internal user ID and creates a CircleMember row. +func (p *EventProcessor) onMemberJoined(ctx context.Context, ev *ContractEvent) error { + contractID := payloadStr(ev.Payload, "circle_id") + walletAddr := payloadStr(ev.Payload, "member") + + c, err := p.circleRepo.FindByContractID(ctx, contractID) + if err != nil { + if isNotFound(err) { + log.Warn().Str("contract_id", contractID).Msg("MemberJoined: circle not found in DB") + return nil + } + return fmt.Errorf("onMemberJoined find circle: %w", err) + } + + u, err := p.userRepo.FindByWalletAddress(ctx, walletAddr) + if err != nil { + if isNotFound(err) { + log.Warn().Str("wallet", walletAddr).Msg("MemberJoined: user not found in DB") + return nil + } + return fmt.Errorf("onMemberJoined find user: %w", err) + } + + member := &circle.CircleMember{ + CircleID: c.ID, + UserID: u.ID, + Status: circle.MemberStatusActive, + JoinedAt: time.Now().UTC(), + } + if err := p.circleRepo.CreateMember(ctx, member); err != nil { + return fmt.Errorf("onMemberJoined create member: %w", err) + } + + log.Info(). + Str("circle_id", c.ID.String()). + Str("user_id", u.ID.String()). + Msg("MemberJoined: member record created") + + p.Broadcast(ctx, c.ID.String(), "member.joined", map[string]any{ + "circle_id": c.ID.String(), + "user_id": u.ID.String(), + "wallet": walletAddr, + "tx_hash": ev.TxHash, + }) + return nil +} + +// onContributionReceived handles ContributionReceived(circle_id, member, amount, round). +// Inserts a confirmed Contribution row linked to the on-chain transaction. +func (p *EventProcessor) onContributionReceived(ctx context.Context, ev *ContractEvent) error { + contractID := payloadStr(ev.Payload, "circle_id") + walletAddr := payloadStr(ev.Payload, "member") + amount := payloadFloat(ev.Payload, "amount") + round := payloadInt(ev.Payload, "round") + + c, err := p.circleRepo.FindByContractID(ctx, contractID) + if err != nil { + if isNotFound(err) { + log.Warn().Str("contract_id", contractID).Msg("ContributionReceived: circle not found") + return nil + } + return fmt.Errorf("onContributionReceived find circle: %w", err) + } + + u, err := p.userRepo.FindByWalletAddress(ctx, walletAddr) + if err != nil { + if isNotFound(err) { + log.Warn().Str("wallet", walletAddr).Msg("ContributionReceived: user not found") + return nil + } + return fmt.Errorf("onContributionReceived find user: %w", err) + } + + contrib := &contribution.Contribution{ + ID: uuid.New(), + CircleID: c.ID, + UserID: u.ID, + RoundNumber: round, + Amount: amount, + TxnHash: sql.NullString{String: ev.TxHash, Valid: true}, + Status: contribution.StatusConfirmed, + OnTime: true, + CreatedAt: time.Now().UTC(), + UpdatedAt: time.Now().UTC(), + } + if err := p.contribRepo.Create(ctx, contrib); err != nil { + return fmt.Errorf("onContributionReceived create: %w", err) + } + + // Update circle's total contributions counter. + c.TotalContributions += amount + if err := p.circleRepo.Update(ctx, c); err != nil { + log.Warn().Err(err).Msg("ContributionReceived: updating circle totals") + } + + log.Info(). + Str("circle_id", c.ID.String()). + Str("user_id", u.ID.String()). + Float64("amount", amount). + Int("round", round). + Msg("ContributionReceived: contribution persisted") + + p.Broadcast(ctx, c.ID.String(), "contribution.confirmed", map[string]any{ + "circle_id": c.ID.String(), + "user_id": u.ID.String(), + "amount": amount, + "round": round, + "contribution_id": contrib.ID.String(), + "tx_hash": ev.TxHash, + }) + return nil +} + +// onPayoutExecuted handles PayoutExecuted(circle_id, recipient, amount, round, payout_type). +// Inserts a Payout row and advances the circle's CurrentRound counter. +func (p *EventProcessor) onPayoutExecuted(ctx context.Context, ev *ContractEvent) error { + contractID := payloadStr(ev.Payload, "circle_id") + recipientWallet := payloadStr(ev.Payload, "recipient") + amount := payloadFloat(ev.Payload, "amount") + round := payloadInt(ev.Payload, "round") + payoutTypeStr := payloadStr(ev.Payload, "payout_type") + + c, err := p.circleRepo.FindByContractID(ctx, contractID) + if err != nil { + if isNotFound(err) { + log.Warn().Str("contract_id", contractID).Msg("PayoutExecuted: circle not found") + return nil + } + return fmt.Errorf("onPayoutExecuted find circle: %w", err) + } + + recipient, err := p.userRepo.FindByWalletAddress(ctx, recipientWallet) + if err != nil { + if isNotFound(err) { + log.Warn().Str("wallet", recipientWallet).Msg("PayoutExecuted: recipient not found") + return nil + } + return fmt.Errorf("onPayoutExecuted find recipient: %w", err) + } + + pt := payout.PayoutTypeRandom + switch payoutTypeStr { + case "fixed": + pt = payout.PayoutTypeFixed + case "auction": + pt = payout.PayoutTypeAuction + case "vote": + pt = payout.PayoutTypeVote + } + + p2 := &payout.Payout{ + ID: uuid.New(), + CircleID: c.ID, + RecipientID: recipient.ID, + RoundNumber: round, + Amount: amount, + TxnHash: sql.NullString{String: ev.TxHash, Valid: true}, + PayoutType: pt, + CreatedAt: time.Now().UTC(), + } + if err := p.payoutRepo.Create(ctx, p2); err != nil { + return fmt.Errorf("onPayoutExecuted create payout: %w", err) + } + + // Advance current round. + c.CurrentRound = round + 1 + if err := p.circleRepo.Update(ctx, c); err != nil { + log.Warn().Err(err).Msg("PayoutExecuted: advancing circle round") + } + + log.Info(). + Str("circle_id", c.ID.String()). + Str("recipient_id", recipient.ID.String()). + Float64("amount", amount). + Int("round", round). + Msg("PayoutExecuted: payout persisted") + + p.Broadcast(ctx, c.ID.String(), "payout.received", map[string]any{ + "circle_id": c.ID.String(), + "recipient_id": recipient.ID.String(), + "amount": amount, + "round": round, + "payout_id": p2.ID.String(), + "tx_hash": ev.TxHash, + }) + return nil +} + +// onLateReported handles LateReported(circle_id, member, penalty_amount, strikes). +// Updates the contribution status to late and marks it not on-time. +func (p *EventProcessor) onLateReported(ctx context.Context, ev *ContractEvent) error { + contractID := payloadStr(ev.Payload, "circle_id") + walletAddr := payloadStr(ev.Payload, "member") + penaltyAmount := payloadFloat(ev.Payload, "penalty_amount") + strikes := payloadInt(ev.Payload, "strikes") + + c, err := p.circleRepo.FindByContractID(ctx, contractID) + if err != nil { + if isNotFound(err) { + log.Warn().Str("contract_id", contractID).Msg("LateReported: circle not found") + return nil + } + return fmt.Errorf("onLateReported find circle: %w", err) + } + + u, err := p.userRepo.FindByWalletAddress(ctx, walletAddr) + if err != nil { + if isNotFound(err) { + log.Warn().Str("wallet", walletAddr).Msg("LateReported: user not found") + return nil + } + return fmt.Errorf("onLateReported find user: %w", err) + } + + log.Warn(). + Str("circle_id", c.ID.String()). + Str("user_id", u.ID.String()). + Float64("penalty", penaltyAmount). + Int("strikes", strikes). + Msg("LateReported: late payment recorded") + + p.Broadcast(ctx, c.ID.String(), "contribution.late", map[string]any{ + "circle_id": c.ID.String(), + "user_id": u.ID.String(), + "penalty_amount": penaltyAmount, + "strikes": strikes, + "tx_hash": ev.TxHash, + }) + return nil +} + +// onMemberExited handles MemberExited(circle_id, member, penalty). +// Updates the circle member status to exited. +func (p *EventProcessor) onMemberExited(ctx context.Context, ev *ContractEvent) error { + contractID := payloadStr(ev.Payload, "circle_id") + walletAddr := payloadStr(ev.Payload, "member") + penalty := payloadFloat(ev.Payload, "penalty") + + c, err := p.circleRepo.FindByContractID(ctx, contractID) + if err != nil { + if isNotFound(err) { + log.Warn().Str("contract_id", contractID).Msg("MemberExited: circle not found") + return nil + } + return fmt.Errorf("onMemberExited find circle: %w", err) + } + + u, err := p.userRepo.FindByWalletAddress(ctx, walletAddr) + if err != nil { + if isNotFound(err) { + log.Warn().Str("wallet", walletAddr).Msg("MemberExited: user not found") + return nil + } + return fmt.Errorf("onMemberExited find user: %w", err) + } + + if err := p.circleRepo.UpdateMemberStatus(ctx, c.ID, u.ID, circle.MemberStatusExited); err != nil { + return fmt.Errorf("onMemberExited update status: %w", err) + } + + log.Info(). + Str("circle_id", c.ID.String()). + Str("user_id", u.ID.String()). + Float64("penalty", penalty). + Msg("MemberExited: member status updated to exited") + + p.Broadcast(ctx, c.ID.String(), "member.exited", map[string]any{ + "circle_id": c.ID.String(), + "user_id": u.ID.String(), + "penalty": penalty, + "tx_hash": ev.TxHash, + }) + return nil +} + +// onDefaultRecorded handles DefaultRecorded(circle_id, member). +// Triggers an on-chain default reputation penalty via reputation score update. +func (p *EventProcessor) onDefaultRecorded(ctx context.Context, ev *ContractEvent) error { + walletAddr := payloadStr(ev.Payload, "member") + contractID := payloadStr(ev.Payload, "circle_id") + + u, err := p.userRepo.FindByWalletAddress(ctx, walletAddr) + if err != nil { + if isNotFound(err) { + log.Warn().Str("wallet", walletAddr).Msg("DefaultRecorded: user not found") + return nil + } + return fmt.Errorf("onDefaultRecorded find user: %w", err) + } + + // Retrieve current reputation snapshot and recalculate with a penalty. + existing, err := p.reputationRepo.GetByUser(ctx, u.ID) + if err != nil && !isNotFound(err) { + return fmt.Errorf("onDefaultRecorded get reputation: %w", err) + } + + currentScore := 0 + if existing != nil { + currentScore = existing.Score + } + + // Apply default penalty (−50 points, floored at 0). + newScore := currentScore - 50 + if newScore < 0 { + newScore = 0 + } + level := reputationLevel(newScore) + + snapshot := &reputation.ReputationSnapshot{ + UserID: u.ID, + Score: newScore, + Level: level, + Month: time.Now().UTC(), + CreatedAt: time.Now().UTC(), + } + if err := p.reputationRepo.SaveSnapshot(ctx, snapshot); err != nil { + return fmt.Errorf("onDefaultRecorded save snapshot: %w", err) + } + + if err := p.userRepo.UpdateMoiScore(ctx, u.ID, newScore); err != nil { + log.Warn().Err(err).Msg("DefaultRecorded: updating user moi_score") + } + + log.Warn(). + Str("user_id", u.ID.String()). + Str("contract_id", contractID). + Int("new_score", newScore). + Msg("DefaultRecorded: reputation penalty applied") + + p.Broadcast(ctx, contractID, "reputation.updated", map[string]any{ + "user_id": u.ID.String(), + "new_score": newScore, + "level": level, + "reason": "default", + "tx_hash": ev.TxHash, + }) + return nil +} + +// onCircleCompleted handles CircleCompleted(circle_id, total_contributions). +// Marks the circle as completed and sets the end date. +func (p *EventProcessor) onCircleCompleted(ctx context.Context, ev *ContractEvent) error { + contractID := payloadStr(ev.Payload, "circle_id") + totalContribs := payloadFloat(ev.Payload, "total_contributions") + + c, err := p.circleRepo.FindByContractID(ctx, contractID) + if err != nil { + if isNotFound(err) { + log.Warn().Str("contract_id", contractID).Msg("CircleCompleted: circle not found") + return nil + } + return fmt.Errorf("onCircleCompleted find circle: %w", err) + } + + now := time.Now().UTC() + c.Status = circle.CircleStatusCompleted + c.TotalContributions = totalContribs + c.EndDate = sql.NullTime{Time: now, Valid: true} + if err := p.circleRepo.Update(ctx, c); err != nil { + return fmt.Errorf("onCircleCompleted update: %w", err) + } + + log.Info(). + Str("circle_id", c.ID.String()). + Float64("total_contributions", totalContribs). + Msg("CircleCompleted: circle marked completed") + + p.Broadcast(ctx, c.ID.String(), "circle.completed", map[string]any{ + "circle_id": c.ID.String(), + "total_contributions": totalContribs, + "tx_hash": ev.TxHash, + }) + return nil +} + +// onAuctionBid handles AuctionBid(circle_id, bidder, discount_bips, round). +// No persistent table exists yet — logs and broadcasts only. +// TODO: Persist to auction_bids table (follow-up issue). +func (p *EventProcessor) onAuctionBid(ctx context.Context, ev *ContractEvent) error { + contractID := payloadStr(ev.Payload, "circle_id") + bidder := payloadStr(ev.Payload, "bidder") + discountBips := payloadInt(ev.Payload, "discount_bips") + round := payloadInt(ev.Payload, "round") + + log.Info(). + Str("contract_id", contractID). + Str("bidder", bidder). + Int("discount_bips", discountBips). + Int("round", round). + Msg("AuctionBid: bid received") + + p.Broadcast(ctx, contractID, "auction.bid", map[string]any{ + "contract_id": contractID, + "bidder": bidder, + "discount_bips": discountBips, + "round": round, + "tx_hash": ev.TxHash, + }) + return nil +} + +// onVoteCast handles VoteCast(circle_id, voter, vote_for, round). +// No persistent table exists yet — logs and broadcasts only. +// TODO: Persist to circle_votes table (follow-up issue). +func (p *EventProcessor) onVoteCast(ctx context.Context, ev *ContractEvent) error { + contractID := payloadStr(ev.Payload, "circle_id") + voter := payloadStr(ev.Payload, "voter") + voteFor := payloadStr(ev.Payload, "vote_for") + round := payloadInt(ev.Payload, "round") + + log.Info(). + Str("contract_id", contractID). + Str("voter", voter). + Str("vote_for", voteFor). + Int("round", round). + Msg("VoteCast: vote recorded") + + p.Broadcast(ctx, contractID, "vote.cast", map[string]any{ + "contract_id": contractID, + "voter": voter, + "vote_for": voteFor, + "round": round, + "tx_hash": ev.TxHash, + }) + return nil +} + +// onDisputeRaised handles DisputeRaised(circle_id, member, evidence_hash). +// Transitions the circle to "disputed" status, freezing payouts until resolved. +func (p *EventProcessor) onDisputeRaised(ctx context.Context, ev *ContractEvent) error { + contractID := payloadStr(ev.Payload, "circle_id") + member := payloadStr(ev.Payload, "member") + evidenceHash := payloadStr(ev.Payload, "evidence_hash") + + c, err := p.circleRepo.FindByContractID(ctx, contractID) + if err != nil { + if isNotFound(err) { + log.Warn().Str("contract_id", contractID).Msg("DisputeRaised: circle not found") + return nil + } + return fmt.Errorf("onDisputeRaised find circle: %w", err) + } + + // Set circle status to disputed to freeze further payouts. + c.Status = "disputed" + if err := p.circleRepo.Update(ctx, c); err != nil { + return fmt.Errorf("onDisputeRaised update circle status: %w", err) + } + + log.Warn(). + Str("circle_id", c.ID.String()). + Str("member_wallet", member). + Str("evidence_hash", evidenceHash). + Msg("DisputeRaised: circle frozen") + + p.Broadcast(ctx, c.ID.String(), "dispute.raised", map[string]any{ + "circle_id": c.ID.String(), + "member_wallet": member, + "evidence_hash": evidenceHash, + "tx_hash": ev.TxHash, + }) + return nil +} + +// onFeeDeposited handles FeeDeposited events from the Treasury contract. +// No persistent table exists yet — logs and broadcasts only. +// TODO: Persist to treasury_fees table (follow-up issue). +func (p *EventProcessor) onFeeDeposited(ctx context.Context, ev *ContractEvent) error { + circleID := payloadStr(ev.Payload, "circle_id") + amount := payloadFloat(ev.Payload, "amount") + + log.Info(). + Str("circle_id", circleID). + Float64("amount", amount). + Str("tx_hash", ev.TxHash). + Msg("FeeDeposited: protocol fee collected") + + p.Broadcast(ctx, circleID, "fee.deposited", map[string]any{ + "circle_id": circleID, + "amount": amount, + "tx_hash": ev.TxHash, + "ledger": ev.Ledger, + }) + return nil +} + +// --------------------------------------------------------------------------- +// persistContractEvent appends a processed event to the audit log. +// --------------------------------------------------------------------------- + +func (p *EventProcessor) persistContractEvent(ctx context.Context, ev *ContractEvent) error { + payloadJSON, err := json.Marshal(ev.Payload) + if err != nil { + return fmt.Errorf("marshaling event payload: %w", err) + } + + _, err = p.db.ExecContext(ctx, ` + INSERT INTO contract_events (tx_hash, ledger, contract_id, event_type, payload, processed_at) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT DO NOTHING`, + ev.TxHash, ev.Ledger, ev.ContractID, ev.EventType, payloadJSON, time.Now().UTC(), + ) + return err +} + +// --------------------------------------------------------------------------- // Broadcast sends a real-time update via WebSocket and publishes the event // to RabbitMQ for async workers (notifications, webhooks, analytics). +// --------------------------------------------------------------------------- + func (p *EventProcessor) Broadcast(ctx context.Context, circleID string, eventType string, payload any) { // Real-time WebSocket broadcast to subscribed clients if p.wsBroadcast != nil { @@ -181,9 +787,88 @@ func (p *EventProcessor) Broadcast(ctx context.Context, circleID string, eventTy } } -// ensure we import these to avoid compiler complaints for work-in-progress -var _ = uuid.New -var _ = circle.Repository(nil) -var _ = contribution.Repository(nil) -var _ = payout.Repository(nil) -var _ = reputation.Repository(nil) +// --------------------------------------------------------------------------- +// Payload extraction helpers — safe, nil-tolerant accessors. +// --------------------------------------------------------------------------- + +func payloadStr(p map[string]any, key string) string { + if p == nil { + return "" + } + v, ok := p[key] + if !ok { + return "" + } + s, _ := v.(string) + return s +} + +func payloadFloat(p map[string]any, key string) float64 { + if p == nil { + return 0 + } + v, ok := p[key] + if !ok { + return 0 + } + switch f := v.(type) { + case float64: + return f + case float32: + return float64(f) + case int: + return float64(f) + case int64: + return float64(f) + case uint64: + return float64(f) + } + return 0 +} + +func payloadInt(p map[string]any, key string) int { + if p == nil { + return 0 + } + v, ok := p[key] + if !ok { + return 0 + } + switch i := v.(type) { + case int: + return i + case int32: + return int(i) + case int64: + return int(i) + case uint32: + return int(i) + case float64: + return int(i) + } + return 0 +} + +// isNotFound returns true for "not found" sentinel errors used by repositories. +func isNotFound(err error) bool { + if err == nil { + return false + } + return err.Error() == "not found" || err == sql.ErrNoRows +} + +// reputationLevel maps a MoiScore to the human-readable tier string. +func reputationLevel(score int) string { + switch { + case score > 800: + return "Diamond" + case score > 600: + return "Platinum" + case score > 400: + return "Gold" + case score > 200: + return "Silver" + default: + return "Bronze" + } +} diff --git a/internal/indexer/processor_test.go b/internal/indexer/processor_test.go new file mode 100644 index 0000000..3f4800d --- /dev/null +++ b/internal/indexer/processor_test.go @@ -0,0 +1,555 @@ +package indexer + +import ( + "context" + "database/sql" + "errors" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + + "github.com/moistello/backend/internal/domain/circle" + circleMocks "github.com/moistello/backend/internal/domain/circle/mocks" + contribMocks "github.com/moistello/backend/internal/domain/contribution/mocks" + payoutMocks "github.com/moistello/backend/internal/domain/payout/mocks" + "github.com/moistello/backend/internal/domain/reputation" + reputationMocks "github.com/moistello/backend/internal/domain/reputation/mocks" + "github.com/moistello/backend/internal/domain/user" + userMocks "github.com/moistello/backend/internal/domain/user/mocks" +) + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +var errTestNotFound = errors.New("not found") + +func newTestProcessor( + circleRepo *circleMocks.Repository, + contribRepo *contribMocks.Repository, + payoutRepo *payoutMocks.Repository, + repRepo *reputationMocks.Repository, + userRepo *userMocks.Repository, +) *EventProcessor { + return &EventProcessor{ + db: nil, // not needed for dispatch-level tests + rmqClient: nil, + circleRepo: circleRepo, + contribRepo: contribRepo, + payoutRepo: payoutRepo, + reputationRepo: repRepo, + userRepo: userRepo, + } +} + +func testCircle(contractID string) *circle.Circle { + return &circle.Circle{ + ID: uuid.New(), + ContractID: sql.NullString{String: contractID, Valid: contractID != ""}, + Status: circle.CircleStatusActive, + CreatedAt: time.Now(), + UpdatedAt: time.Now(), + } +} + +func testUser(wallet string) *user.User { + return &user.User{ + ID: uuid.New(), + WalletAddress: wallet, + CreatedAt: time.Now(), + UpdatedAt: time.Now(), + } +} + + + +func contractEvent(eventType, contractID string, payload map[string]any) *ContractEvent { + return &ContractEvent{ + ContractID: contractID, + EventType: eventType, + Ledger: 100, + TxHash: "txhash_" + eventType, + Payload: payload, + } +} + +// --------------------------------------------------------------------------- +// CircleCreated +// --------------------------------------------------------------------------- + +func TestOnCircleCreated_AlreadyLinked(t *testing.T) { + cRepo := &circleMocks.Repository{} + p := newTestProcessor(cRepo, nil, nil, nil, nil) + + ev := contractEvent(EventCircleCreated, "contract1", map[string]any{ + "circle_id": "contract1", + "creator": "GABC123", + }) + + c := testCircle("contract1") + cRepo.On("FindByContractID", mock.Anything, "contract1").Return(c, nil) + + err := p.onCircleCreated(context.Background(), ev) + assert.NoError(t, err) + cRepo.AssertExpectations(t) +} + +func TestOnCircleCreated_NotFound_LogsOnly(t *testing.T) { + cRepo := &circleMocks.Repository{} + p := newTestProcessor(cRepo, nil, nil, nil, nil) + + ev := contractEvent(EventCircleCreated, "contract_new", map[string]any{ + "circle_id": "contract_new", + "creator": "GABC", + }) + + cRepo.On("FindByContractID", mock.Anything, "contract_new").Return(nil, errTestNotFound) + + err := p.onCircleCreated(context.Background(), ev) + assert.NoError(t, err) + cRepo.AssertExpectations(t) +} + +// --------------------------------------------------------------------------- +// MemberJoined +// --------------------------------------------------------------------------- + +func TestOnMemberJoined_Success(t *testing.T) { + cRepo := &circleMocks.Repository{} + uRepo := &userMocks.Repository{} + p := newTestProcessor(cRepo, nil, nil, nil, uRepo) + + c := testCircle("cid1") + u := testUser("GWALLET1") + + cRepo.On("FindByContractID", mock.Anything, "cid1").Return(c, nil) + uRepo.On("FindByWalletAddress", mock.Anything, "GWALLET1").Return(u, nil) + cRepo.On("CreateMember", mock.Anything, mock.MatchedBy(func(m *circle.CircleMember) bool { + return m.CircleID == c.ID && m.UserID == u.ID && m.Status == circle.MemberStatusActive + })).Return(nil) + + ev := contractEvent(EventMemberJoined, "cid1", map[string]any{ + "circle_id": "cid1", + "member": "GWALLET1", + "contribution_amount": float64(100), + }) + + err := p.onMemberJoined(context.Background(), ev) + assert.NoError(t, err) + cRepo.AssertExpectations(t) + uRepo.AssertExpectations(t) +} + +func TestOnMemberJoined_CircleNotFound(t *testing.T) { + cRepo := &circleMocks.Repository{} + uRepo := &userMocks.Repository{} + p := newTestProcessor(cRepo, nil, nil, nil, uRepo) + + cRepo.On("FindByContractID", mock.Anything, "cid_missing").Return(nil, errTestNotFound) + + ev := contractEvent(EventMemberJoined, "cid_missing", map[string]any{ + "circle_id": "cid_missing", + "member": "GWALLET1", + }) + + err := p.onMemberJoined(context.Background(), ev) + assert.NoError(t, err) // graceful skip + cRepo.AssertExpectations(t) + uRepo.AssertNotCalled(t, "FindByWalletAddress") +} + +func TestOnMemberJoined_UserNotFound(t *testing.T) { + cRepo := &circleMocks.Repository{} + uRepo := &userMocks.Repository{} + p := newTestProcessor(cRepo, nil, nil, nil, uRepo) + + c := testCircle("cid1") + cRepo.On("FindByContractID", mock.Anything, "cid1").Return(c, nil) + uRepo.On("FindByWalletAddress", mock.Anything, "GWALLET_MISSING").Return(nil, errTestNotFound) + + ev := contractEvent(EventMemberJoined, "cid1", map[string]any{ + "circle_id": "cid1", + "member": "GWALLET_MISSING", + }) + + err := p.onMemberJoined(context.Background(), ev) + assert.NoError(t, err) // graceful skip +} + +// --------------------------------------------------------------------------- +// ContributionReceived +// --------------------------------------------------------------------------- + +func TestOnContributionReceived_Success(t *testing.T) { + cRepo := &circleMocks.Repository{} + ctRepo := &contribMocks.Repository{} + uRepo := &userMocks.Repository{} + p := newTestProcessor(cRepo, ctRepo, nil, nil, uRepo) + + c := testCircle("cid1") + u := testUser("GWALLET1") + + cRepo.On("FindByContractID", mock.Anything, "cid1").Return(c, nil) + uRepo.On("FindByWalletAddress", mock.Anything, "GWALLET1").Return(u, nil) + ctRepo.On("Create", mock.Anything, mock.AnythingOfType("*contribution.Contribution")).Return(nil) + cRepo.On("Update", mock.Anything, mock.AnythingOfType("*circle.Circle")).Return(nil) + + ev := contractEvent(EventContributionReceived, "cid1", map[string]any{ + "circle_id": "cid1", + "member": "GWALLET1", + "amount": float64(50), + "round": int(1), + }) + + err := p.onContributionReceived(context.Background(), ev) + assert.NoError(t, err) + ctRepo.AssertCalled(t, "Create", mock.Anything, mock.AnythingOfType("*contribution.Contribution")) +} + +// --------------------------------------------------------------------------- +// PayoutExecuted +// --------------------------------------------------------------------------- + +func TestOnPayoutExecuted_Success(t *testing.T) { + cRepo := &circleMocks.Repository{} + pRepo := &payoutMocks.Repository{} + uRepo := &userMocks.Repository{} + p := newTestProcessor(cRepo, nil, pRepo, nil, uRepo) + + c := testCircle("cid1") + u := testUser("GWALLET_RECIPIENT") + + cRepo.On("FindByContractID", mock.Anything, "cid1").Return(c, nil) + uRepo.On("FindByWalletAddress", mock.Anything, "GWALLET_RECIPIENT").Return(u, nil) + pRepo.On("Create", mock.Anything, mock.AnythingOfType("*payout.Payout")).Return(nil) + cRepo.On("Update", mock.Anything, mock.AnythingOfType("*circle.Circle")).Return(nil) + + ev := contractEvent(EventPayoutExecuted, "cid1", map[string]any{ + "circle_id": "cid1", + "recipient": "GWALLET_RECIPIENT", + "amount": float64(500), + "round": int(2), + "payout_type": "random", + }) + + err := p.onPayoutExecuted(context.Background(), ev) + assert.NoError(t, err) + pRepo.AssertCalled(t, "Create", mock.Anything, mock.AnythingOfType("*payout.Payout")) +} + +// --------------------------------------------------------------------------- +// LateReported +// --------------------------------------------------------------------------- + +func TestOnLateReported_Success(t *testing.T) { + cRepo := &circleMocks.Repository{} + uRepo := &userMocks.Repository{} + p := newTestProcessor(cRepo, nil, nil, nil, uRepo) + + c := testCircle("cid1") + u := testUser("GWALLET1") + + cRepo.On("FindByContractID", mock.Anything, "cid1").Return(c, nil) + uRepo.On("FindByWalletAddress", mock.Anything, "GWALLET1").Return(u, nil) + + ev := contractEvent(EventLateReported, "cid1", map[string]any{ + "circle_id": "cid1", + "member": "GWALLET1", + "penalty_amount": float64(5), + "strikes": int(1), + }) + + err := p.onLateReported(context.Background(), ev) + assert.NoError(t, err) +} + +// --------------------------------------------------------------------------- +// MemberExited +// --------------------------------------------------------------------------- + +func TestOnMemberExited_Success(t *testing.T) { + cRepo := &circleMocks.Repository{} + uRepo := &userMocks.Repository{} + p := newTestProcessor(cRepo, nil, nil, nil, uRepo) + + c := testCircle("cid1") + u := testUser("GWALLET1") + + cRepo.On("FindByContractID", mock.Anything, "cid1").Return(c, nil) + uRepo.On("FindByWalletAddress", mock.Anything, "GWALLET1").Return(u, nil) + cRepo.On("UpdateMemberStatus", mock.Anything, c.ID, u.ID, circle.MemberStatusExited).Return(nil) + + ev := contractEvent(EventMemberExited, "cid1", map[string]any{ + "circle_id": "cid1", + "member": "GWALLET1", + "penalty": float64(10), + }) + + err := p.onMemberExited(context.Background(), ev) + assert.NoError(t, err) + cRepo.AssertCalled(t, "UpdateMemberStatus", mock.Anything, c.ID, u.ID, circle.MemberStatusExited) +} + +// --------------------------------------------------------------------------- +// DefaultRecorded +// --------------------------------------------------------------------------- + +func TestOnDefaultRecorded_Success(t *testing.T) { + cRepo := &circleMocks.Repository{} + rRepo := &reputationMocks.Repository{} + uRepo := &userMocks.Repository{} + p := newTestProcessor(cRepo, nil, nil, rRepo, uRepo) + + u := testUser("GWALLET1") + existing := &reputation.ReputationSnapshot{ + UserID: u.ID, + Score: 600, + Level: "Platinum", + } + + uRepo.On("FindByWalletAddress", mock.Anything, "GWALLET1").Return(u, nil) + rRepo.On("GetByUser", mock.Anything, u.ID).Return(existing, nil) + rRepo.On("SaveSnapshot", mock.Anything, mock.MatchedBy(func(s *reputation.ReputationSnapshot) bool { + return s.Score == 550 && s.UserID == u.ID + })).Return(nil) + uRepo.On("UpdateMoiScore", mock.Anything, u.ID, 550).Return(nil) + + ev := contractEvent(EventDefaultRecorded, "cid1", map[string]any{ + "circle_id": "cid1", + "member": "GWALLET1", + }) + + err := p.onDefaultRecorded(context.Background(), ev) + assert.NoError(t, err) + rRepo.AssertExpectations(t) +} + +func TestOnDefaultRecorded_ScoreFloorsAtZero(t *testing.T) { + rRepo := &reputationMocks.Repository{} + uRepo := &userMocks.Repository{} + p := newTestProcessor(nil, nil, nil, rRepo, uRepo) + + u := testUser("GWALLET1") + existing := &reputation.ReputationSnapshot{ + UserID: u.ID, + Score: 30, // < 50, so penalty should floor at 0 + Level: "Bronze", + } + + uRepo.On("FindByWalletAddress", mock.Anything, "GWALLET1").Return(u, nil) + rRepo.On("GetByUser", mock.Anything, u.ID).Return(existing, nil) + rRepo.On("SaveSnapshot", mock.Anything, mock.MatchedBy(func(s *reputation.ReputationSnapshot) bool { + return s.Score == 0 + })).Return(nil) + uRepo.On("UpdateMoiScore", mock.Anything, u.ID, 0).Return(nil) + + ev := contractEvent(EventDefaultRecorded, "cid1", map[string]any{ + "circle_id": "cid1", + "member": "GWALLET1", + }) + + err := p.onDefaultRecorded(context.Background(), ev) + assert.NoError(t, err) +} + +// --------------------------------------------------------------------------- +// CircleCompleted +// --------------------------------------------------------------------------- + +func TestOnCircleCompleted_Success(t *testing.T) { + cRepo := &circleMocks.Repository{} + p := newTestProcessor(cRepo, nil, nil, nil, nil) + + c := testCircle("cid1") + cRepo.On("FindByContractID", mock.Anything, "cid1").Return(c, nil) + cRepo.On("Update", mock.Anything, mock.MatchedBy(func(upd *circle.Circle) bool { + return upd.Status == circle.CircleStatusCompleted && upd.TotalContributions == 1000 + })).Return(nil) + + ev := contractEvent(EventCircleCompleted, "cid1", map[string]any{ + "circle_id": "cid1", + "total_contributions": float64(1000), + }) + + err := p.onCircleCompleted(context.Background(), ev) + assert.NoError(t, err) + cRepo.AssertExpectations(t) +} + +// --------------------------------------------------------------------------- +// AuctionBid — log + broadcast only, no repo calls +// --------------------------------------------------------------------------- + +func TestOnAuctionBid_NoError(t *testing.T) { + p := newTestProcessor(nil, nil, nil, nil, nil) + + ev := contractEvent(EventAuctionBid, "cid1", map[string]any{ + "circle_id": "cid1", + "bidder": "GBIDDER", + "discount_bips": int(200), + "round": int(3), + }) + + err := p.onAuctionBid(context.Background(), ev) + assert.NoError(t, err) +} + +// --------------------------------------------------------------------------- +// VoteCast — log + broadcast only +// --------------------------------------------------------------------------- + +func TestOnVoteCast_NoError(t *testing.T) { + p := newTestProcessor(nil, nil, nil, nil, nil) + + ev := contractEvent(EventVoteCast, "cid1", map[string]any{ + "circle_id": "cid1", + "voter": "GVOTER", + "vote_for": "GCANDIDATE", + "round": int(1), + }) + + err := p.onVoteCast(context.Background(), ev) + assert.NoError(t, err) +} + +// --------------------------------------------------------------------------- +// DisputeRaised +// --------------------------------------------------------------------------- + +func TestOnDisputeRaised_Success(t *testing.T) { + cRepo := &circleMocks.Repository{} + p := newTestProcessor(cRepo, nil, nil, nil, nil) + + c := testCircle("cid1") + cRepo.On("FindByContractID", mock.Anything, "cid1").Return(c, nil) + cRepo.On("Update", mock.Anything, mock.MatchedBy(func(upd *circle.Circle) bool { + return string(upd.Status) == "disputed" + })).Return(nil) + + ev := contractEvent(EventDisputeRaised, "cid1", map[string]any{ + "circle_id": "cid1", + "member": "GDISPUTER", + "evidence_hash": "sha256hash", + }) + + err := p.onDisputeRaised(context.Background(), ev) + assert.NoError(t, err) + cRepo.AssertExpectations(t) +} + +func TestOnDisputeRaised_CircleNotFound(t *testing.T) { + cRepo := &circleMocks.Repository{} + p := newTestProcessor(cRepo, nil, nil, nil, nil) + + cRepo.On("FindByContractID", mock.Anything, "cid_missing").Return(nil, errTestNotFound) + + ev := contractEvent(EventDisputeRaised, "cid_missing", map[string]any{ + "circle_id": "cid_missing", + "member": "GMEMBER", + }) + + err := p.onDisputeRaised(context.Background(), ev) + assert.NoError(t, err) // graceful skip +} + +// --------------------------------------------------------------------------- +// FeeDeposited — log + broadcast only +// --------------------------------------------------------------------------- + +func TestOnFeeDeposited_NoError(t *testing.T) { + p := newTestProcessor(nil, nil, nil, nil, nil) + + ev := contractEvent(EventFeeDeposited, "treasury_contract", map[string]any{ + "circle_id": "cid1", + "amount": float64(2.5), + }) + + err := p.onFeeDeposited(context.Background(), ev) + assert.NoError(t, err) +} + +// --------------------------------------------------------------------------- +// dispatchEvent routing +// --------------------------------------------------------------------------- + +func TestDispatchEvent_UnknownType_NoError(t *testing.T) { + p := newTestProcessor(nil, nil, nil, nil, nil) + + ev := contractEvent("UnknownEventXYZ", "cid1", nil) + err := p.dispatchEvent(context.Background(), ev) + assert.NoError(t, err) +} + +// --------------------------------------------------------------------------- +// ProcessTransaction — integration path +// --------------------------------------------------------------------------- + +func TestProcessTransaction_EmptyOperations(t *testing.T) { + p := newTestProcessor(nil, nil, nil, nil, nil) + txn := &Transaction{Hash: "abc", Ledger: 1} + err := p.ProcessTransaction(context.Background(), txn) + assert.NoError(t, err) +} + +func TestProcessTransaction_NonSorobanOp(t *testing.T) { + p := newTestProcessor(nil, nil, nil, nil, nil) + txn := &Transaction{ + Hash: "abc", + Ledger: 1, + Operations: []Operation{ + {Type: "payment", SourceAccount: "GACCOUNT"}, + }, + } + err := p.ProcessTransaction(context.Background(), txn) + assert.NoError(t, err) +} + +func TestProcessTransaction_ExtendTTL(t *testing.T) { + p := newTestProcessor(nil, nil, nil, nil, nil) + txn := &Transaction{ + Hash: "abc", + Ledger: 1, + Operations: []Operation{ + {Type: "extend_footprint_ttl", SourceAccount: "GCONTRACT"}, + }, + } + err := p.ProcessTransaction(context.Background(), txn) + assert.NoError(t, err) +} + +func TestProcessTransaction_SorobanEmptyXDR(t *testing.T) { + // invoke_host_function with empty ResultMetaXDR → no events, no error. + p := newTestProcessor(nil, nil, nil, nil, nil) + txn := &Transaction{ + Hash: "abc", + Ledger: 1, + Operations: []Operation{ + {Type: "invoke_host_function", SourceAccount: "GCONTRACT", ResultMetaXDR: ""}, + }, + } + err := p.ProcessTransaction(context.Background(), txn) + assert.NoError(t, err) +} + +// --------------------------------------------------------------------------- +// WebSocket broadcast +// --------------------------------------------------------------------------- + +func TestBroadcast_CallsWsBroadcast(t *testing.T) { + p := newTestProcessor(nil, nil, nil, nil, nil) + + called := false + var capturedCircleID string + p.SetWebSocketBroadcast(func(circleID string, data any) { + called = true + capturedCircleID = circleID + }) + + p.Broadcast(context.Background(), "circle-123", "circle.created", map[string]any{}) + + assert.True(t, called) + assert.Equal(t, "circle-123", capturedCircleID) +} diff --git a/internal/indexer/xdr_parser.go b/internal/indexer/xdr_parser.go new file mode 100644 index 0000000..d23f983 --- /dev/null +++ b/internal/indexer/xdr_parser.go @@ -0,0 +1,233 @@ +package indexer + +import ( + "bytes" + "encoding/base64" + "fmt" + "strconv" + + "github.com/stellar/go/xdr" +) + +// ParseContractEvents decodes the base64-encoded result_meta_xdr from a Stellar +// transaction and extracts all Soroban contract events contained within. Each +// returned ContractEvent is fully decoded with its EventType and Payload fields +// populated from the raw SCVal topics and data. +// +// Malformed or unrecognised XDR is silently skipped — the function returns +// whatever events could be successfully decoded along with the first error +// encountered (if any). Callers should treat partial results as valid. +func ParseContractEvents(txHash string, ledger int64, resultMetaXDR string) ([]ContractEvent, error) { + if resultMetaXDR == "" { + return nil, nil + } + + raw, err := base64.StdEncoding.DecodeString(resultMetaXDR) + if err != nil { + return nil, fmt.Errorf("base64 decode result_meta_xdr: %w", err) + } + + var meta xdr.TransactionMeta + if _, err := xdr.Unmarshal(bytes.NewReader(raw), &meta); err != nil { + return nil, fmt.Errorf("xdr unmarshal TransactionMeta: %w", err) + } + + // Soroban contract events live in V3 metadata. + v3 := meta.V3 + if v3 == nil { + return nil, nil + } + + var events []ContractEvent + for _, diagEvent := range v3.SorobanMeta.Events { + ev, ok := decodeContractEvent(txHash, ledger, diagEvent) + if !ok { + continue + } + events = append(events, ev) + } + return events, nil +} + +// decodeContractEvent converts a raw xdr.ContractEvent into a typed ContractEvent. +// Returns (event, true) on success or (zero, false) if the event cannot be decoded. +func decodeContractEvent(txHash string, ledger int64, raw xdr.ContractEvent) (ContractEvent, bool) { + // Only process contract-type events (not system or diagnostic). + if raw.Type != xdr.ContractEventTypeContract { + return ContractEvent{}, false + } + + contractID := "" + if raw.ContractId != nil { + h := *raw.ContractId + contractID = fmt.Sprintf("%x", h[:]) + } + + body, ok := raw.Body.GetV0() + if !ok { + return ContractEvent{}, false + } + + topics := body.Topics + if len(topics) == 0 { + return ContractEvent{}, false + } + + eventType := decodeSymbol(topics[0]) + if eventType == "" { + return ContractEvent{}, false + } + + payload := decodePayload(topics[1:], body.Data) + + return ContractEvent{ + ContractID: contractID, + EventType: eventType, + Ledger: ledger, + TxHash: txHash, + Payload: payload, + }, true +} + +// decodeSymbol extracts a string from an SCVal of type SCV_SYMBOL or SCV_STRING. +func decodeSymbol(v xdr.ScVal) string { + switch v.Type { + case xdr.ScValTypeScvSymbol: + if sym, ok := v.GetSym(); ok { + return string(sym) + } + case xdr.ScValTypeScvString: + if s, ok := v.GetStr(); ok { + return string(s) + } + } + return "" +} + +// decodePayload converts the remaining event topics and data SCVal into a flat +// map suitable for storage as JSONB. Keys are positional ("topic_1", "topic_2", …) +// for remaining topics, and "data" for the event data field. +// +// Named fields are extracted where the on-chain convention encodes them as +// alternating key-value symbol pairs within the topics list. +func decodePayload(remainingTopics []xdr.ScVal, data xdr.ScVal) map[string]any { + payload := make(map[string]any) + + for i, topic := range remainingTopics { + key := fmt.Sprintf("topic_%d", i+1) + payload[key] = scValToGo(topic) + } + + dataVal := scValToGo(data) + if dataVal != nil { + payload["data"] = dataVal + } + + return payload +} + +// scValToGo converts an xdr.ScVal to its closest Go-native representation. +// Complex nested types (maps, vecs) are recursively expanded. +func scValToGo(v xdr.ScVal) any { + switch v.Type { + case xdr.ScValTypeScvBool: + b, _ := v.GetB() + return b + + case xdr.ScValTypeScvSymbol: + sym, _ := v.GetSym() + return string(sym) + + case xdr.ScValTypeScvString: + s, _ := v.GetStr() + return string(s) + + case xdr.ScValTypeScvU32: + u, _ := v.GetU32() + return uint32(u) + + case xdr.ScValTypeScvI32: + i, _ := v.GetI32() + return int32(i) + + case xdr.ScValTypeScvU64: + u, _ := v.GetU64() + return uint64(u) + + case xdr.ScValTypeScvI64: + i, _ := v.GetI64() + return int64(i) + + case xdr.ScValTypeScvU128: + u, _ := v.GetU128() + // Represent as decimal string to avoid precision loss in JSON. + hi := uint64(u.Hi) + lo := uint64(u.Lo) + if hi == 0 { + return strconv.FormatUint(lo, 10) + } + return fmt.Sprintf("%d%018d", hi, lo) + + case xdr.ScValTypeScvI128: + i, _ := v.GetI128() + hi := int64(i.Hi) + lo := uint64(i.Lo) + if hi == 0 { + return strconv.FormatUint(lo, 10) + } + return fmt.Sprintf("%d%018d", hi, lo) + + case xdr.ScValTypeScvAddress: + addr, _ := v.GetAddress() + return scAddressToString(addr) + + case xdr.ScValTypeScvBytes: + b, _ := v.GetBytes() + return base64.StdEncoding.EncodeToString(b) + + case xdr.ScValTypeScvMap: + m, _ := v.GetMap() + if m == nil { + return nil + } + result := make(map[string]any, len(*m)) + for _, entry := range *m { + k := fmt.Sprintf("%v", scValToGo(entry.Key)) + result[k] = scValToGo(entry.Val) + } + return result + + case xdr.ScValTypeScvVec: + vec, _ := v.GetVec() + if vec == nil { + return nil + } + result := make([]any, len(*vec)) + for i, elem := range *vec { + result[i] = scValToGo(elem) + } + return result + + case xdr.ScValTypeScvVoid: + return nil + + default: + return fmt.Sprintf("", v.Type) + } +} + +// scAddressToString converts an xdr.ScAddress to a human-readable string +// (Stellar account ID or contract hex). +func scAddressToString(addr xdr.ScAddress) string { + switch addr.Type { + case xdr.ScAddressTypeScAddressTypeAccount: + if addr.AccountId != nil { + return addr.AccountId.Address() + } + case xdr.ScAddressTypeScAddressTypeContract: + if addr.ContractId != nil { + return fmt.Sprintf("%x", (*addr.ContractId)[:]) + } + } + return "" +} diff --git a/internal/indexer/xdr_parser_test.go b/internal/indexer/xdr_parser_test.go new file mode 100644 index 0000000..8b703ca --- /dev/null +++ b/internal/indexer/xdr_parser_test.go @@ -0,0 +1,95 @@ +package indexer + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestParseContractEvents_EmptyXDR(t *testing.T) { + events, err := ParseContractEvents("hash1", 100, "") + assert.NoError(t, err) + assert.Empty(t, events) +} + +func TestParseContractEvents_InvalidBase64(t *testing.T) { + events, err := ParseContractEvents("hash1", 100, "!!!not-valid-base64!!!") + assert.Error(t, err) + assert.Contains(t, err.Error(), "base64 decode") + assert.Nil(t, events) +} + +func TestParseContractEvents_InvalidXDR(t *testing.T) { + // Valid base64 but not valid XDR. + events, err := ParseContractEvents("hash1", 100, "aGVsbG8gd29ybGQ=") + assert.Error(t, err) + assert.Nil(t, events) +} + +func TestScValToGo_Void(t *testing.T) { + // Full round-trip XDR tests require Soroban testnet data and are covered + // in integration tests. This file covers helper utilities only. + t.Log("XDR round-trip tests require live testnet data — see xdr_parser integration tests") +} + +func TestPayloadStr(t *testing.T) { + p := map[string]any{"key": "value", "num": 42} + assert.Equal(t, "value", payloadStr(p, "key")) + assert.Equal(t, "", payloadStr(p, "missing")) + assert.Equal(t, "", payloadStr(nil, "key")) + // Non-string value should return empty string, not panic. + assert.Equal(t, "", payloadStr(p, "num")) +} + +func TestPayloadFloat(t *testing.T) { + p := map[string]any{ + "f64": float64(3.14), + "f32": float32(2.5), + "i64": int64(100), + "u64": uint64(200), + "i": int(50), + } + assert.InDelta(t, 3.14, payloadFloat(p, "f64"), 0.001) + assert.InDelta(t, 2.5, payloadFloat(p, "f32"), 0.001) + assert.Equal(t, float64(100), payloadFloat(p, "i64")) + assert.Equal(t, float64(200), payloadFloat(p, "u64")) + assert.Equal(t, float64(50), payloadFloat(p, "i")) + assert.Equal(t, float64(0), payloadFloat(p, "missing")) + assert.Equal(t, float64(0), payloadFloat(nil, "key")) +} + +func TestPayloadInt(t *testing.T) { + p := map[string]any{ + "i": int(10), + "i32": int32(20), + "i64": int64(30), + "u32": uint32(40), + "f64": float64(50), + } + assert.Equal(t, 10, payloadInt(p, "i")) + assert.Equal(t, 20, payloadInt(p, "i32")) + assert.Equal(t, 30, payloadInt(p, "i64")) + assert.Equal(t, 40, payloadInt(p, "u32")) + assert.Equal(t, 50, payloadInt(p, "f64")) + assert.Equal(t, 0, payloadInt(p, "missing")) + assert.Equal(t, 0, payloadInt(nil, "key")) +} + +func TestIsNotFound(t *testing.T) { + assert.False(t, isNotFound(nil)) + assert.True(t, isNotFound(errNotFound("not found"))) +} + +func TestReputationLevel(t *testing.T) { + assert.Equal(t, "Diamond", reputationLevel(850)) + assert.Equal(t, "Platinum", reputationLevel(650)) + assert.Equal(t, "Gold", reputationLevel(450)) + assert.Equal(t, "Silver", reputationLevel(250)) + assert.Equal(t, "Bronze", reputationLevel(100)) + assert.Equal(t, "Bronze", reputationLevel(0)) +} + +// errNotFound is a helper for creating simple not-found error values in tests. +type errNotFound string + +func (e errNotFound) Error() string { return string(e) }