Architecture of Chaos Part 3 — Event Sourcing Saved Our Audit Trail, Then a Fiber Cable Broke

State(t) = Reduce(all_events_up_to_t, initial_state)

-- migrations/001_event_store.sql
CREATE TABLE event_store (
    event_id        UUID PRIMARY KEY,
    event_type      VARCHAR(64) NOT NULL,
    aggregate_id    UUID NOT NULL,
    stream_version  BIGINT NOT NULL,
    causation_id    UUID,
    correlation_id  UUID NOT NULL,
    hlc_physical    BIGINT NOT NULL,
    hlc_logical     INTEGER NOT NULL,
    vector_clock    JSONB NOT NULL,
    payload         JSONB NOT NULL,
    metadata        JSONB NOT NULL,
    user_id         UUID,
    recorded_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    UNIQUE (aggregate_id, stream_version)
);

-- Append-only protection: ban UPDATE and DELETE
CREATE RULE no_update AS ON UPDATE TO event_store DO INSTEAD NOTHING;
CREATE RULE no_delete AS ON DELETE TO event_store DO INSTEAD NOTHING;

// commands/place_bid.ts
class PlaceBidHandler {
  async handle(command: PlaceBidCommand): Promise<CommandResult> {
    // 1. Idempotency check
    const existing = await this.eventStore.findByCorrelationId(command.commandId);
    if (existing.length > 0) {
      return { status: 'already_processed', events: existing };
    }

    // 2. Load aggregates
    const userBalance = await this.userBalanceProjection.get(command.bidderId);
    const auction = await this.auctionProjection.get(command.auctionId);

    // 3. Business rules (invariants)
    if (auction.status !== 'ACTIVE')
      return { status: 'rejected', reason: 'Auction not active' };
    if (command.amount <= auction.currentHighestBid)
      return { status: 'rejected', reason: 'Bid too low' };
    if (userBalance.availableBalance < command.amount)
      return { status: 'rejected', reason: 'Insufficient funds' };

    // 4. Produce events atomically
    const events = [
      { eventType: 'BID_PLACED', aggregateId: command.auctionId, ... },
      { eventType: 'FUNDS_RESERVED', aggregateId: command.bidderId, ... },
    ];

    await this.eventStore.append(events);
    await this.eventBus.publish(events);
    return { status: 'accepted', events };
  }
}

// workflows/auction_win.go — Temporal Workflow
func AuctionWinWorkflow(ctx workflow.Context, input AuctionWinInput) error {
    retryPolicy := &temporal.RetryPolicy{
        InitialInterval:    time.Second,
        BackoffCoefficient: 2.0,
        MaximumAttempts:    5,
    }

    ctx = workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
        StartToCloseTimeout: 30 * time.Second,
        RetryPolicy:         retryPolicy,
    })

    // STEP 1: Reserve funds
    var reservationID string
    err := workflow.ExecuteActivity(ctx, ReserveFunds, input).Get(ctx, &reservationID)
    if err != nil { return err }

    // STEP 2: Transfer to escrow
    var escrowID string
    err = workflow.ExecuteActivity(ctx, TransferToEscrow, input).Get(ctx, &escrowID)
    if err != nil {
        // COMPENSATE: Release reservation
        workflow.ExecuteActivity(ctx, ReleaseReservation, reservationID).Get(ctx, nil)
        return err
    }

    // STEP 3: Credit seller (pending)
    err = workflow.ExecuteActivity(ctx, CreditSellerPending, input).Get(ctx, nil)
    if err != nil {
        // COMPENSATE: Refund escrow + release reservation
        workflow.ExecuteActivity(ctx, RefundFromEscrow, escrowID).Get(ctx, nil)
        workflow.ExecuteActivity(ctx, ReleaseReservation, reservationID).Get(ctx, nil)
        return err
    }

    // STEP 4: Transfer ownership
    err = workflow.ExecuteActivity(ctx, TransferOwnership, input).Get(ctx, nil)
    if err != nil {
        // COMPENSATE: All steps in reverse
        workflow.ExecuteActivity(ctx, DebitSellerPending, input).Get(ctx, nil)
        workflow.ExecuteActivity(ctx, RefundFromEscrow, escrowID).Get(ctx, nil)
        workflow.ExecuteActivity(ctx, ReleaseReservation, reservationID).Get(ctx, nil)
        return err
    }

    // STEP 5: Email (best-effort, failure won't fail the workflow)
    _ = workflow.ExecuteActivity(ctx, SendWinnerEmail, input).Get(ctx, nil)

    return nil
}

func ReserveFunds(ctx context.Context, input ReserveFundsInput) (string, error) {
    // Idempotency key from input hash
    key := generateIdempotencyKey(input.UserID, input.Amount, input.CorrelationID)

    existing, err := db.GetReservationByIdempotencyKey(ctx, key)
    if err == nil && existing != nil {
        return existing.ReservationID, nil  // Already processed
    }

    // Create new reservation in a transaction
    reservationID := ulid.Make().String()
    err = db.WithTransaction(ctx, func(tx *sql.Tx) error {
        result, err := tx.ExecContext(ctx, `
            UPDATE user_balances
            SET available_balance = available_balance - $1,
                reserved_balance = reserved_balance + $1
            WHERE user_id = $2 AND available_balance >= $1
        `, input.Amount, input.UserID)
        if err != nil { return err }
        if rows, _ := result.RowsAffected(); rows == 0 {
            return ErrInsufficientFunds
        }
        _, err = tx.ExecContext(ctx, `
            INSERT INTO reservations (reservation_id, user_id, amount, idempotency_key, status)
            VALUES ($1, $2, $3, $4, 'ACTIVE')
        `, reservationID, input.UserID, input.Amount, key)
        return err
    })
    return reservationID, err
}