feat: wire outbox worker in financial-accounting service (#422)

* feat: wire outbox worker in financial-accounting service

Integrate the transactional outbox pattern worker into the financial-accounting
service for reliable event publishing with at-least-once delivery guarantees.

Changes:
- Add ControlAction type and ControlLog domain method for SUSPEND/RESUME/TERMINATE
  lifecycle operations following BIAN CoCR patterns
- Add WithTransaction and DB methods to LedgerRepository for atomic outbox writes
- Add protobuf definitions for ControlFinancialBookingLog RPC and events
- Wire OutboxRepository and Worker in main.go with Kafka producer
- Implement ControlFinancialBookingLog gRPC method with idempotency support
- Add comprehensive tests for control operations and state machine logic

The outbox worker polls for pending events and publishes to Kafka, with
configurable batch size and poll interval. Events are written atomically
with domain state changes in a single transaction.

* fix: address CodeRabbit review feedback

Addresses three critical issues identified by CodeRabbit:

1. Kafka producer flush: Call Flush() with 5s timeout before Close() to
   ensure all pending outbox events are delivered before shutdown.

2. Race condition fix: Eliminate double-fetch by performing all operations
   within a single transaction with pessimistic locking (SELECT FOR UPDATE).
   Previously fetched entity outside transaction for domain logic, then
   again inside transaction, creating window for concurrent modifications.
   Now fetch-lock-apply-save happens atomically.

3. Layer separation: Service layer reconstructs domain model from locked
   entity, applies domain logic, then updates entity - maintaining proper
   separation while avoiding race conditions.

The refactored ControlBookingLog method now:
- Acquires pessimistic lock on entity within transaction
- Reconstructs domain model from locked entity
- Applies domain control logic
- Updates locked entity with domain results
- Writes event to outbox atomically
- Maps all domain errors to gRPC codes after transaction

---------

Co-authored-by: Ben Coombs <bjcoombs@users.noreply.github.com>

Author: bjcoombs

api/proto/meridian/events/v1/financial_accounting_events.proto

@@ -452,6 +452,65 @@ message LedgerPostingRejectedEvent {
   int64 version = 8 [(buf.validate.field).int64 = {gte: 1}];
 }
 
+// FinancialBookingLogControlledEvent represents a control action applied to
+// a financial booking log (SUSPEND, RESUME, TERMINATE). Published when
+// administrative control operations are performed.
+message FinancialBookingLogControlledEvent {
+  // booking_log_id uniquely identifies the financial booking log
+  string booking_log_id = 1 [(buf.validate.field).string = {
+    min_len: 1
+    max_len: 255
+  }];
+
+  // control_action is the action that was performed (SUSPEND, RESUME, TERMINATE)
+  string control_action = 2 [(buf.validate.field).string = {
+    min_len: 1
+    max_len: 50
+  }];
+
+  // previous_status is the status before the control action
+  meridian.common.v1.TransactionStatus previous_status = 3 [(buf.validate.field).enum = {
+    defined_only: true
+    not_in: [0]
+  }];
+
+  // new_status is the status after the control action
+  meridian.common.v1.TransactionStatus new_status = 4 [(buf.validate.field).enum = {
+    defined_only: true
+    not_in: [0]
+  }];
+
+  // reason explains why the control action was performed
+  string reason = 5 [(buf.validate.field).string = {
+    min_len: 1
+    max_len: 500
+  }];
+
+  // controlled_by identifies who performed the control action
+  string controlled_by = 6 [(buf.validate.field).string = {
+    min_len: 1
+    max_len: 100
+  }];
+
+  // correlation_id links related events across services
+  string correlation_id = 7 [(buf.validate.field).string = {
+    min_len: 1
+    max_len: 255
+  }];
+
+  // causation_id identifies the event or command that caused this event
+  string causation_id = 8 [(buf.validate.field).string = {
+    min_len: 1
+    max_len: 255
+  }];
+
+  // timestamp when the event was created
+  google.protobuf.Timestamp timestamp = 9 [(buf.validate.field).required = true];
+
+  // version is the aggregate version for optimistic locking
+  int64 version = 10 [(buf.validate.field).int64 = {gte: 1}];
+}
+
 // BalanceValidationFailedEvent represents a booking log that failed
 // double-entry balance validation (debits != credits). Published when
 // attempting to post an unbalanced booking log.

api/proto/meridian/financial_accounting/v1/financial_accounting.proto

@@ -104,6 +104,19 @@ message FinancialBookingLog {
   repeated LedgerPosting postings = 10;
 }
 
+// ControlAction defines the lifecycle control actions for financial booking logs.
+// These are BIAN control operations for administrative management of booking logs.
+enum ControlAction {
+  // CONTROL_ACTION_UNSPECIFIED means the action is unknown.
+  CONTROL_ACTION_UNSPECIFIED = 0;
+  // CONTROL_ACTION_SUSPEND temporarily suspends the booking log (transitions to FAILED/suspended).
+  CONTROL_ACTION_SUSPEND = 1;
+  // CONTROL_ACTION_RESUME reactivates a suspended booking log (returns to PENDING).
+  CONTROL_ACTION_RESUME = 2;
+  // CONTROL_ACTION_TERMINATE permanently ends the booking log lifecycle (transitions to CANCELLED).
+  CONTROL_ACTION_TERMINATE = 3;
+}
+
 // LedgerPosting represents a single posting operation in double-entry bookkeeping.
 message LedgerPosting {
   // id is the unique identifier for this posting.
@@ -429,6 +442,44 @@ message ListLedgerPostingsResponse {
   meridian.common.v1.PaginationResponse pagination = 2;
 }
 
+// ControlFinancialBookingLogRequest controls the lifecycle of a booking log (BIAN CoCR).
+//
+// This operation supports administrative control actions:
+//   - SUSPEND: Temporarily suspend processing of the booking log
+//   - RESUME: Resume processing of a previously suspended booking log
+//   - TERMINATE: Permanently terminate the booking log lifecycle
+//
+// The operation uses the transactional outbox pattern to ensure atomic
+// persistence of both the state change and the associated event.
+message ControlFinancialBookingLogRequest {
+  // id is the booking log to control.
+  string id = 1 [(buf.validate.field).string = {
+    min_len: 1
+    max_len: 255
+  }];
+
+  // control_action is the lifecycle action to perform.
+  ControlAction control_action = 2 [(buf.validate.field).enum = {
+    defined_only: true
+    not_in: [0] // Prevent UNSPECIFIED
+  }];
+
+  // reason is an explanation for the control action (required for audit trail).
+  string reason = 3 [(buf.validate.field).string = {
+    min_len: 1
+    max_len: 500
+  }];
+
+  // idempotency_key ensures exactly-once processing.
+  meridian.common.v1.IdempotencyKey idempotency_key = 4 [(buf.validate.field).required = true];
+}
+
+// ControlFinancialBookingLogResponse returns the controlled booking log.
+message ControlFinancialBookingLogResponse {
+  // financial_booking_log is the booking log after the control action.
+  FinancialBookingLog financial_booking_log = 1 [(buf.validate.field).required = true];
+}
+
 // FinancialAccountingService provides the BIAN Financial Accounting service interface.
 service FinancialAccountingService {
   // InitiateFinancialBookingLog creates a new financial booking log.
@@ -506,4 +557,14 @@ service FinancialAccountingService {
       get: "/v1/postings"
     };
   }
+
+  // ControlFinancialBookingLog controls the lifecycle of a booking log (BIAN CoCR).
+  // Supports SUSPEND, RESUME, and TERMINATE control actions.
+  // Uses transactional outbox pattern for reliable event publishing.
+  rpc ControlFinancialBookingLog(ControlFinancialBookingLogRequest) returns (ControlFinancialBookingLogResponse) {
+    option (google.api.http) = {
+      post: "/v1/booking-logs/{id}/control"
+      body: "*"
+    };
+  }
 }

services/financial-accounting/adapters/persistence/repository.go

@@ -666,3 +666,29 @@ func (r *LedgerRepository) ListPostings(ctx context.Context, params ListPostings
 	}
 	return result, nil
 }
+
+// WithTransaction executes a function within a database transaction with tenant scoping.
+// This is used for implementing the transactional outbox pattern where both the entity
+// update and the outbox event write must succeed or fail together atomically.
+//
+// The provided function receives a tenant-scoped *gorm.DB transaction that can be used
+// for all database operations within the transaction.
+//
+// Example:
+//
+//	err := repo.WithTransaction(ctx, func(tx *gorm.DB) error {
+//	    if err := tx.Save(&entity).Error; err != nil {
+//	        return err
+//	    }
+//	    return outboxRepo.Insert(ctx, tx, event)
+//	})
+func (r *LedgerRepository) WithTransaction(ctx context.Context, fn func(tx *gorm.DB) error) error {
+	return r.withTenantTransaction(ctx, fn)
+}
+
+// DB returns the underlying GORM database instance.
+// This is primarily used for passing the DB to other components that need
+// database access, such as the outbox repository for the transactional outbox pattern.
+func (r *LedgerRepository) DB() *gorm.DB {
+	return r.db
+}

services/financial-accounting/cmd/main.go

@@ -14,6 +14,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
 	financialaccountingv1 "github.com/meridianhub/meridian/api/proto/meridian/financial_accounting/v1"
 	"github.com/meridianhub/meridian/services/financial-accounting/adapters/persistence"
 	serviceobs "github.com/meridianhub/meridian/services/financial-accounting/observability"
@@ -22,6 +23,7 @@ import (
 	"github.com/meridianhub/meridian/shared/pkg/interceptors"
 	"github.com/meridianhub/meridian/shared/platform/audit"
 	"github.com/meridianhub/meridian/shared/platform/auth"
+	"github.com/meridianhub/meridian/shared/platform/events"
 	"github.com/meridianhub/meridian/shared/platform/observability"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/redis/go-redis/v9"
@@ -39,6 +41,11 @@ var (
 	BuildDate = "unknown"
 )
 
+// Package-level variables for lifecycle management
+var (
+	outboxWorker *events.Worker
+)
+
 // Static errors for configuration validation
 var (
 	ErrBankCashAccountIDRequired = errors.New("BANK_CASH_ACCOUNT_ID environment variable is required")
@@ -126,6 +133,47 @@ func run(logger *slog.Logger) error {
 		}()
 	}
 
+	// Initialize outbox repository and worker for transactional event publishing.
+	// The outbox pattern ensures at-least-once delivery of events by storing them
+	// in the database first and then publishing asynchronously via Kafka.
+	outboxRepo := events.NewPostgresOutboxRepository(db)
+	outboxPublisher := events.NewOutboxPublisher("financial-accounting")
+
+	// Initialize Kafka producer for outbox worker (optional - depends on KAFKA_BOOTSTRAP_SERVERS)
+	var kafkaProducer *kafka.Producer
+	bootstrapServers := getEnvOrDefault("KAFKA_BOOTSTRAP_SERVERS", "")
+	if bootstrapServers != "" {
+		producer, err := kafka.NewProducer(&kafka.ConfigMap{
+			"bootstrap.servers": bootstrapServers,
+			"client.id":         "financial-accounting-outbox-worker",
+			"acks":              "all",
+			"retries":           3,
+			"compression.type":  "snappy",
+			"linger.ms":         10,
+			"batch.size":        16384,
+		})
+		if err != nil {
+			logger.Warn("failed to create Kafka producer for outbox worker",
+				"error", err)
+		} else {
+			kafkaProducer = producer
+			logger.Info("Kafka producer initialized for outbox worker",
+				"bootstrap_servers", bootstrapServers)
+		}
+	} else {
+		logger.Info("outbox worker disabled: KAFKA_BOOTSTRAP_SERVERS not set")
+	}
+
+	// Start outbox worker if Kafka producer is available
+	if kafkaProducer != nil {
+		workerConfig := events.DefaultWorkerConfig("financial-accounting")
+		outboxWorker = events.NewWorker(outboxRepo, kafkaProducer, workerConfig, logger)
+		outboxWorker.Start(ctx)
+		logger.Info("outbox worker started",
+			"batch_size", workerConfig.BatchSize,
+			"poll_interval", workerConfig.PollInterval)
+	}
+
 	// Validate bank cash account ID is configured
 	bankCashAccountID := getEnvOrDefault("BANK_CASH_ACCOUNT_ID", "")
 	if bankCashAccountID == "" {
@@ -171,7 +219,13 @@ func run(logger *slog.Logger) error {
 	logger.Info("event publisher initialized (noop mode)")
 
 	// Create Financial Accounting service
-	financialAccountingSvc, err := service.NewFinancialAccountingService(ledgerRepo, eventPublisher, idempotencySvc)
+	financialAccountingSvc, err := service.NewFinancialAccountingService(
+		ledgerRepo,
+		eventPublisher,
+		idempotencySvc,
+		outboxPublisher,
+		outboxRepo,
+	)
 	if err != nil {
 		return fmt.Errorf("failed to create financial accounting service: %w", err)
 	}
@@ -314,7 +368,24 @@ func run(logger *slog.Logger) error {
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
-	// Shutdown HTTP server first (faster, allows metrics scraping during gRPC drain)
+	// Stop outbox worker first (stop processing before closing Kafka producer)
+	if outboxWorker != nil {
+		logger.Info("stopping outbox worker...")
+		outboxWorker.Stop()
+		logger.Info("outbox worker stopped")
+	}
+
+	// Close Kafka producer after outbox worker stops
+	if kafkaProducer != nil {
+		logger.Info("flushing Kafka producer...")
+		// Flush pending messages with 5 second timeout to ensure delivery
+		kafkaProducer.Flush(5000) // 5 seconds in milliseconds
+		logger.Info("closing Kafka producer...")
+		kafkaProducer.Close()
+		logger.Info("Kafka producer closed")
+	}
+
+	// Shutdown HTTP server (faster, allows metrics scraping during gRPC drain)
 	if err := httpServer.Shutdown(shutdownCtx); err != nil {
 		logger.Error("HTTP server shutdown error", "error", err)
 	} else {