feat: add idempotency protection to update operations (#379)

Author: bjcoombs

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

@@ -220,6 +220,9 @@ message UpdateFinancialBookingLogRequest {
   // If empty, the existing rules are preserved (optional update).
   // Validation: Min length is enforced only if the field is provided (service layer).
   string chart_of_accounts_rules = 3;
+
+  // idempotency_key ensures exactly-once processing.
+  meridian.common.v1.IdempotencyKey idempotency_key = 4 [(buf.validate.field).required = true];
 }
 
 // UpdateFinancialBookingLogResponse returns the updated booking log.
@@ -316,6 +319,9 @@ message UpdateLedgerPostingRequest {
   // posting_result describes the outcome of the posting operation.
   // If empty, the existing result is preserved (optional update).
   string posting_result = 3 [(buf.validate.field).string = {max_len: 1000}];
+
+  // idempotency_key ensures exactly-once processing.
+  meridian.common.v1.IdempotencyKey idempotency_key = 4 [(buf.validate.field).required = true];
 }
 
 // UpdateLedgerPostingResponse returns the updated posting.

services/financial-accounting/service/financial_accounting_service.go

@@ -543,23 +543,77 @@ func (s *FinancialAccountingService) ListLedgerPostings(
 // UpdateLedgerPosting updates an existing ledger posting's status and result.
 //
 // Workflow:
-// 1. Parse and validate request fields
-// 2. Retrieve existing posting by ID
-// 3. Validate state transition rules (e.g., cannot change POSTED status)
-// 4. Apply update using domain methods (Post/Fail)
-// 5. Persist updated posting
-// 6. Publish domain event (LedgerPostingUpdatedEvent)
-// 7. Return updated posting
+// 1. Check idempotency using request's IdempotencyKey
+// 2. Parse and validate request fields
+// 3. Retrieve existing posting by ID
+// 4. Validate state transition rules (e.g., cannot change POSTED status)
+// 5. Apply update using domain methods (Post/Fail)
+// 6. Persist updated posting
+// 7. Publish domain event (LedgerPostingUpdatedEvent)
+// 8. Return updated posting
+//
+// Idempotency Note:
+// Unlike CaptureLedgerPosting where idempotency is optional (create operations
+// naturally fail on duplicate IDs), update operations REQUIRE idempotency keys
+// because state-machine transitions must be exactly-once. A duplicate update
+// could incorrectly transition an entity through multiple states.
 //
 // Error mapping:
 // - Invalid request fields -> codes.InvalidArgument
+// - Duplicate idempotency key -> codes.AlreadyExists
 // - Posting not found -> codes.NotFound
 // - Invalid state transition -> codes.FailedPrecondition
 // - Internal errors -> codes.Internal
 func (s *FinancialAccountingService) UpdateLedgerPosting(
 	ctx context.Context,
 	req *financialaccountingv1.UpdateLedgerPostingRequest,
 ) (*financialaccountingv1.UpdateLedgerPostingResponse, error) {
+	// Validate idempotency key is provided
+	if req.IdempotencyKey == nil || req.IdempotencyKey.Key == "" {
+		return nil, status.Error(codes.InvalidArgument, "idempotency_key is required")
+	}
+
+	idempotencyKey := idempotency.Key{
+		Namespace: "financial-accounting",
+		Operation: "update-posting",
+		EntityID:  req.GetId(),
+		RequestID: req.IdempotencyKey.Key,
+	}
+
+	// Check idempotency - defensive nil check (constructor requires non-nil service)
+	// TODO(ledger-integrity#15): If an error occurs after MarkPending but before StoreResult,
+	// the pending marker is left orphaned until TTL expiry. Consider adding cleanup on error.
+	if s.idempotency != nil {
+		result, err := s.idempotency.Check(ctx, idempotencyKey)
+		if err != nil && !errors.Is(err, idempotency.ErrResultNotFound) {
+			if errors.Is(err, idempotency.ErrOperationAlreadyProcessed) {
+				if result != nil && result.Status == idempotency.StatusCompleted && len(result.Data) > 0 {
+					// Deserialize cached response from protobuf
+					var cachedResponse financialaccountingv1.UpdateLedgerPostingResponse
+					if unmarshalErr := proto.Unmarshal(result.Data, &cachedResponse); unmarshalErr != nil {
+						slog.Error("failed to deserialize cached idempotency response",
+							"error", unmarshalErr,
+							"idempotency_key", req.IdempotencyKey.Key,
+							"operation", "update-posting")
+						return nil, status.Error(codes.AlreadyExists, "request with this idempotency key already processed")
+					}
+					slog.Info("returning cached idempotent response",
+						"idempotency_key", req.IdempotencyKey.Key,
+						"operation", "update-posting",
+						"posting_id", req.GetId())
+					return &cachedResponse, nil
+				}
+				return nil, status.Error(codes.AlreadyExists, "request with this idempotency key already processed")
+			}
+			return nil, status.Errorf(codes.Internal, "failed to check idempotency: %v", err)
+		}
+
+		// Mark as pending to prevent concurrent processing
+		if err := s.idempotency.MarkPending(ctx, idempotencyKey, defaultIdempotencyTTL); err != nil {
+			return nil, status.Errorf(codes.Internal, "failed to mark operation as pending: %v", err)
+		}
+	}
+
 	// Parse posting ID
 	postingID, err := parseUUID(req.GetId())
 	if err != nil {
@@ -636,9 +690,43 @@ func (s *FinancialAccountingService) UpdateLedgerPosting(
 	// TODO(75-async-audit#5): Implement LedgerPostingUpdatedEvent and publish it
 
 	// Convert to proto response
-	return &financialaccountingv1.UpdateLedgerPostingResponse{
+	response := &financialaccountingv1.UpdateLedgerPostingResponse{
 		LedgerPosting: toProtoLedgerPosting(posting),
-	}, nil
+	}
+
+	// Store idempotency result (only if service configured)
+	if s.idempotency != nil {
+		ttl := defaultIdempotencyTTL
+		if req.IdempotencyKey.TtlSeconds > 0 {
+			ttl = time.Duration(req.IdempotencyKey.TtlSeconds) * time.Second
+		}
+
+		// Serialize response using protobuf for idempotent storage
+		responseData, marshalErr := proto.Marshal(response)
+		if marshalErr != nil {
+			slog.Error("failed to serialize response for idempotency cache",
+				"error", marshalErr,
+				"idempotency_key", req.IdempotencyKey.Key,
+				"operation", "update-posting")
+		} else {
+			result := idempotency.Result{
+				Key:         idempotencyKey,
+				Status:      idempotency.StatusCompleted,
+				Data:        responseData,
+				CompletedAt: time.Now(),
+				TTL:         ttl,
+			}
+
+			if storeErr := s.idempotency.StoreResult(ctx, result); storeErr != nil {
+				slog.Error("failed to store idempotency result",
+					"error", storeErr,
+					"idempotency_key", req.IdempotencyKey.Key,
+					"operation", "update-posting")
+			}
+		}
+	}
+
+	return response, nil
 }
 
 // isValidCurrencyCode validates that a currency code matches ISO 4217 format.
@@ -810,22 +898,76 @@ func (s *FinancialAccountingService) RetrieveFinancialBookingLog(
 // UpdateFinancialBookingLog updates an existing booking log's status and rules.
 //
 // Workflow:
-// 1. Parse and validate request fields
-// 2. Retrieve existing booking log by ID
-// 3. Validate state transition rules
-// 4. Apply updates using domain methods
-// 5. Persist updated booking log
-// 6. Return updated booking log
+// 1. Check idempotency using request's IdempotencyKey
+// 2. Parse and validate request fields
+// 3. Retrieve existing booking log by ID
+// 4. Validate state transition rules
+// 5. Apply updates using domain methods
+// 6. Persist updated booking log
+// 7. Return updated booking log
+//
+// Idempotency Note:
+// Unlike InitiateFinancialBookingLog where idempotency is optional (create operations
+// naturally fail on duplicate IDs), update operations REQUIRE idempotency keys
+// because state-machine transitions must be exactly-once. A duplicate update
+// could incorrectly transition an entity through multiple states.
 //
 // Error mapping:
 // - Invalid request fields -> codes.InvalidArgument
+// - Duplicate idempotency key -> codes.AlreadyExists
 // - Booking log not found -> codes.NotFound
 // - Invalid state transition -> codes.FailedPrecondition
 // - Internal errors -> codes.Internal
 func (s *FinancialAccountingService) UpdateFinancialBookingLog(
 	ctx context.Context,
 	req *financialaccountingv1.UpdateFinancialBookingLogRequest,
 ) (*financialaccountingv1.UpdateFinancialBookingLogResponse, error) {
+	// Validate idempotency key is provided
+	if req.IdempotencyKey == nil || req.IdempotencyKey.Key == "" {
+		return nil, status.Error(codes.InvalidArgument, "idempotency_key is required")
+	}
+
+	idempotencyKey := idempotency.Key{
+		Namespace: "financial-accounting",
+		Operation: "update-booking-log",
+		EntityID:  req.GetId(),
+		RequestID: req.IdempotencyKey.Key,
+	}
+
+	// Check idempotency - defensive nil check (constructor requires non-nil service)
+	// TODO(ledger-integrity#15): If an error occurs after MarkPending but before StoreResult,
+	// the pending marker is left orphaned until TTL expiry. Consider adding cleanup on error.
+	if s.idempotency != nil {
+		result, err := s.idempotency.Check(ctx, idempotencyKey)
+		if err != nil && !errors.Is(err, idempotency.ErrResultNotFound) {
+			if errors.Is(err, idempotency.ErrOperationAlreadyProcessed) {
+				if result != nil && result.Status == idempotency.StatusCompleted && len(result.Data) > 0 {
+					// Deserialize cached response from protobuf
+					var cachedResponse financialaccountingv1.UpdateFinancialBookingLogResponse
+					if unmarshalErr := proto.Unmarshal(result.Data, &cachedResponse); unmarshalErr != nil {
+						slog.Error("failed to deserialize cached idempotency response",
+							"error", unmarshalErr,
+							"idempotency_key", req.IdempotencyKey.Key,
+							"operation", "update-booking-log")
+						return nil, status.Error(codes.AlreadyExists, "request with this idempotency key already processed")
+					}
+					slog.Info("returning cached idempotent response",
+						"idempotency_key", req.IdempotencyKey.Key,
+						"operation", "update-booking-log",
+						"booking_log_id", req.GetId())
+					return &cachedResponse, nil
+				}
+				return nil, status.Error(codes.AlreadyExists, "request with this idempotency key already processed")
+			}
+			return nil, status.Errorf(codes.Internal, "failed to check idempotency: %v", err)
+		}
+
+		// Mark as pending to prevent concurrent processing
+		if err := s.idempotency.MarkPending(ctx, idempotencyKey, defaultIdempotencyTTL); err != nil {
+			return nil, status.Errorf(codes.Internal, "failed to mark operation as pending: %v", err)
+		}
+	}
+
 	// Parse booking log ID
 	bookingLogID, err := parseUUID(req.GetId())
 	if err != nil {
@@ -935,9 +1077,43 @@ func (s *FinancialAccountingService) UpdateFinancialBookingLog(
 	// TODO(75-async-audit#5): Publish FinancialBookingLogUpdatedEvent for inter-service coordination
 
 	// Convert to proto response
-	return &financialaccountingv1.UpdateFinancialBookingLogResponse{
+	response := &financialaccountingv1.UpdateFinancialBookingLogResponse{
 		FinancialBookingLog: toProtoFinancialBookingLog(&updated),
-	}, nil
+	}
+
+	// Store idempotency result (only if service configured)
+	if s.idempotency != nil {
+		ttl := defaultIdempotencyTTL
+		if req.IdempotencyKey.TtlSeconds > 0 {
+			ttl = time.Duration(req.IdempotencyKey.TtlSeconds) * time.Second
+		}
+
+		// Serialize response using protobuf for idempotent storage
+		responseData, marshalErr := proto.Marshal(response)
+		if marshalErr != nil {
+			slog.Error("failed to serialize response for idempotency cache",
+				"error", marshalErr,
+				"idempotency_key", req.IdempotencyKey.Key,
+				"operation", "update-booking-log")
+		} else {
+			result := idempotency.Result{
+				Key:         idempotencyKey,
+				Status:      idempotency.StatusCompleted,
+				Data:        responseData,
+				CompletedAt: time.Now(),
+				TTL:         ttl,
+			}
+
+			if storeErr := s.idempotency.StoreResult(ctx, result); storeErr != nil {
+				slog.Error("failed to store idempotency result",
+					"error", storeErr,
+					"idempotency_key", req.IdempotencyKey.Key,
+					"operation", "update-booking-log")
+			}
+		}
+	}
+
+	return response, nil
 }
 
 // isValidBookingLogTransition validates that a status transition is allowed.