package postgres import ( "context" "errors" "strings" "testing" "time" "gl/domain/ledger" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" ) type fakeRow struct { values []any err error } func (r fakeRow) Scan(dest ...any) error { if r.err != nil { return r.err } if len(dest) != len(r.values) { return errors.New("unexpected scan width") } for index, value := range r.values { switch target := dest[index].(type) { case *string: *target = value.(string) case *int64: *target = value.(int64) case *bool: *target = value.(bool) case *uint32: *target = value.(uint32) case *time.Time: *target = value.(time.Time) case *ledger.TransactionState: *target = value.(ledger.TransactionState) case *[]byte: *target = value.([]byte) default: return errors.New("unsupported scan target") } } return nil } type fakeTx struct { rows []Row execCount int execErrAt int committed bool rolledBack bool } func (t *fakeTx) Exec(context.Context, string, ...any) (pgconn.CommandTag, error) { t.execCount++ if t.execCount == t.execErrAt { return pgconn.CommandTag{}, errors.New("exec failed") } return pgconn.NewCommandTag("INSERT 0 1"), nil } func (t *fakeTx) QueryRow(context.Context, string, ...any) Row { row := t.rows[0] t.rows = t.rows[1:] return row } func (t *fakeTx) Commit(context.Context) error { t.committed = true return nil } func (t *fakeTx) Rollback(context.Context) error { t.rolledBack = true return nil } type fakeDatabase struct { tx *fakeTx directRow Row directRows []Row beginCalled bool } func (d *fakeDatabase) Begin(context.Context) (Tx, error) { d.beginCalled = true return d.tx, nil } func (d *fakeDatabase) QueryRow(context.Context, string, ...any) Row { if len(d.directRows) > 0 { row := d.directRows[0] d.directRows = d.directRows[1:] return row } return d.directRow } func (d *fakeDatabase) Query(context.Context, string, ...any) (Rows, error) { return nil, errors.New("unexpected query") } func (d *fakeDatabase) Exec(context.Context, string, ...any) (pgconn.CommandTag, error) { return pgconn.CommandTag{}, errors.New("unexpected exec") } func (d *fakeDatabase) Ping(context.Context) error { return nil } func (d *fakeDatabase) Close() {} func TestJournalRepositoryAppendCommitsJournalEntriesAndSeal(t *testing.T) { journal := repositoryJournal(t) tx := &fakeTx{rows: []Row{ fakeRow{values: []any{journal.ID}}, fakeRow{values: []any{int64(10)}}, fakeRow{values: []any{int64(20)}}, }} database := &fakeDatabase{tx: tx} result, err := NewJournalRepository(database).Append(context.Background(), journal) if err != nil { t.Fatal(err) } if result.JournalID != journal.ID || result.AlreadyExists { t.Fatalf("unexpected result: %+v", result) } if !tx.committed || tx.rolledBack { t.Fatalf("unexpected transaction state: %+v", tx) } if tx.execCount != 3 { t.Fatalf("expected two entry inserts and one seal, got %d execs", tx.execCount) } } func TestJournalRepositoryReturnsExistingIdenticalJournal(t *testing.T) { journal := repositoryJournal(t) tx := &fakeTx{rows: []Row{fakeRow{err: pgx.ErrNoRows}}} database := &fakeDatabase{ tx: tx, directRow: fakeRow{values: []any{journal.ID, journal.PayloadHash, true}}, } result, err := NewJournalRepository(database).Append(context.Background(), journal) if err != nil { t.Fatal(err) } if !result.AlreadyExists || result.JournalID != journal.ID { t.Fatalf("unexpected result: %+v", result) } if !tx.rolledBack || tx.committed { t.Fatalf("duplicate transaction was not rolled back: %+v", tx) } } func TestJournalRepositoryRejectsConflictingIdempotencyPayload(t *testing.T) { journal := repositoryJournal(t) database := &fakeDatabase{ tx: &fakeTx{rows: []Row{fakeRow{err: pgx.ErrNoRows}}}, directRow: fakeRow{values: []any{journal.ID, strings.Repeat("b", 64), true}}, } _, err := NewJournalRepository(database).Append(context.Background(), journal) if !errors.Is(err, ErrIdempotencyConflict) { t.Fatalf("expected idempotency conflict, got %v", err) } } func TestJournalRepositoryRollsBackEntryFailure(t *testing.T) { journal := repositoryJournal(t) tx := &fakeTx{ rows: []Row{ fakeRow{values: []any{journal.ID}}, fakeRow{values: []any{int64(10)}}, }, execErrAt: 1, } database := &fakeDatabase{tx: tx} if _, err := NewJournalRepository(database).Append(context.Background(), journal); err == nil { t.Fatal("expected entry insert error") } if !tx.rolledBack || tx.committed { t.Fatalf("failed append was not rolled back: %+v", tx) } } func TestJournalRepositoryValidatesBeforeOpeningTransaction(t *testing.T) { journal := repositoryJournal(t) journal.Entries[1].Amount, _ = ledger.ParseAmount("9") database := &fakeDatabase{tx: &fakeTx{}} if _, err := NewJournalRepository(database).Append(context.Background(), journal); err == nil { t.Fatal("expected validation error") } if database.beginCalled { t.Fatal("invalid journal opened a database transaction") } } func repositoryJournal(t *testing.T) ledger.Journal { t.Helper() debit, err := ledger.ParseAmount("-10.25") if err != nil { t.Fatal(err) } return ledger.Journal{ ID: "11111111-1111-4111-8111-111111111111", SourceService: "wallet", IdempotencyKey: "wallet:1:internal-transfer:v1", SourceTransactionID: "1", TrackingCode: "track-1", EffectKind: "internal-transfer", EventVersion: 1, OccurredAt: time.Unix(1, 0).UTC(), PayloadHash: strings.Repeat("a", 64), Metadata: map[string]string{"origin": "test"}, Entries: []ledger.Entry{ { LineNumber: 1, Account: ledger.AccountReference{ Class: ledger.AccountClassUserAvailable, OwnerType: "user", OwnerID: "1", AssetID: 5, }, Amount: debit, }, { LineNumber: 2, Account: ledger.AccountReference{ Class: ledger.AccountClassUserAvailable, OwnerType: "user", OwnerID: "2", AssetID: 5, }, Amount: debit.Negate(), }, }, } }