GL/infrastructure/postgres/journal_repository.go

package postgres

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"

	"gl/domain/ledger"

	"github.com/jackc/pgx/v5"
)

var (
	ErrIdempotencyConflict = errors.New("idempotency key belongs to a different payload")
	ErrIncompleteJournal   = errors.New("idempotency key belongs to an unsealed journal")
)

type AppendResult struct {
	JournalID     string
	AlreadyExists bool
}

type JournalRepository struct {
	database Database
}

func NewJournalRepository(database Database) *JournalRepository {
	return &JournalRepository{database: database}
}

func (r *JournalRepository) Append(ctx context.Context, journal ledger.Journal) (result AppendResult, err error) {
	if err := journal.Validate(); err != nil {
		return AppendResult{}, fmt.Errorf("validate journal: %w", err)
	}
	metadataValues := journal.Metadata
	if metadataValues == nil {
		metadataValues = map[string]string{}
	}
	metadata, err := json.Marshal(metadataValues)
	if err != nil {
		return AppendResult{}, fmt.Errorf("encode metadata: %w", err)
	}

	tx, err := r.database.Begin(ctx)
	if err != nil {
		return AppendResult{}, fmt.Errorf("begin journal append: %w", err)
	}
	defer func() {
		if err != nil {
			_ = tx.Rollback(ctx)
		}
	}()

	var insertedID string
	err = tx.QueryRow(ctx, insertJournalSQL,
		journal.ID,
		journal.SourceService,
		journal.IdempotencyKey,
		journal.SourceTransactionID,
		journal.TrackingCode,
		journal.EffectKind,
		journal.EventVersion,
		nullIfEmpty(journal.ReversalOfJournalID),
		journal.OccurredAt,
		journal.CorrelationID,
		journal.ActorID,
		journal.Blockchain.Network,
		journal.Blockchain.TransactionHash,
		journal.Blockchain.LedgerSequence,
		metadata,
		journal.PayloadHash,
	).Scan(&insertedID)
	if errors.Is(err, pgx.ErrNoRows) {
		_ = tx.Rollback(ctx)
		return r.resolveDuplicate(ctx, journal)
	}
	if err != nil {
		return AppendResult{}, fmt.Errorf("insert journal: %w", err)
	}

	for _, entry := range journal.Entries {
		var accountID int64
		err = tx.QueryRow(ctx, insertAccountSQL,
			entry.Account.Class,
			entry.Account.OwnerType,
			entry.Account.OwnerID,
			entry.Account.AssetID,
		).Scan(&accountID)
		if errors.Is(err, pgx.ErrNoRows) {
			err = tx.QueryRow(ctx, selectAccountSQL,
				entry.Account.Class,
				entry.Account.OwnerType,
				entry.Account.OwnerID,
				entry.Account.AssetID,
			).Scan(&accountID)
		}
		if err != nil {
			return AppendResult{}, fmt.Errorf("resolve entry %d account: %w", entry.LineNumber, err)
		}

		if _, err = tx.Exec(ctx, insertEntrySQL,
			journal.ID,
			entry.LineNumber,
			accountID,
			entry.Account.AssetID,
			entry.Amount.String(),
			entry.Description,
		); err != nil {
			return AppendResult{}, fmt.Errorf("insert entry %d: %w", entry.LineNumber, err)
		}
	}

	if _, err = tx.Exec(ctx, "UPDATE journals SET sealed_at = clock_timestamp() WHERE id = $1", journal.ID); err != nil {
		return AppendResult{}, fmt.Errorf("seal journal: %w", err)
	}
	if err = tx.Commit(ctx); err != nil {
		return AppendResult{}, fmt.Errorf("commit journal: %w", err)
	}
	return AppendResult{JournalID: insertedID}, nil
}

func (r *JournalRepository) resolveDuplicate(ctx context.Context, journal ledger.Journal) (AppendResult, error) {
	var (
		journalID   string
		payloadHash string
		sealed      bool
	)
	err := r.database.QueryRow(ctx, selectIdempotencySQL, journal.IdempotencyKey).Scan(&journalID, &payloadHash, &sealed)
	if err != nil {
		return AppendResult{}, fmt.Errorf("read idempotent journal: %w", err)
	}
	if payloadHash != journal.PayloadHash {
		return AppendResult{}, ErrIdempotencyConflict
	}
	if !sealed {
		return AppendResult{}, ErrIncompleteJournal
	}
	return AppendResult{JournalID: journalID, AlreadyExists: true}, nil
}

func nullIfEmpty(value string) any {
	if value == "" {
		return nil
	}
	return value
}

const insertJournalSQL = `
INSERT INTO journals (
    id, source_service, idempotency_key, source_transaction_id, tracking_code,
    effect_kind, event_version, reversal_of_journal_id, occurred_at,
    correlation_id, actor_id, blockchain_network, blockchain_transaction_hash,
    blockchain_ledger_sequence, metadata, payload_hash
) VALUES (
    $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16
)
ON CONFLICT (idempotency_key) DO NOTHING
RETURNING id`

const insertAccountSQL = `
INSERT INTO ledger_accounts (class, owner_type, owner_id, asset_id)
VALUES ($1, $2, $3, $4)
ON CONFLICT (class, owner_type, owner_id, asset_id) DO NOTHING
RETURNING id`

const selectAccountSQL = `
SELECT id FROM ledger_accounts
WHERE class = $1 AND owner_type = $2 AND owner_id = $3 AND asset_id = $4`

const insertEntrySQL = `
INSERT INTO journal_entries (
    journal_id, line_number, account_id, asset_id, amount, description
) VALUES ($1, $2, $3, $4, $5, $6)`

const selectIdempotencySQL = `
SELECT id, payload_hash, sealed_at IS NOT NULL
FROM journals
WHERE idempotency_key = $1`