Method RunExport

svc/ctrl/worker/auditlogexport/run_export_handler.go:51–88

RunExport drains the clickhouse_outbox table into ClickHouse in batches. Each batch is its own restate.Run so a crash mid-drain only replays the last incomplete batch. Within a batch: 1. SELECT outbox rows WHERE deleted_at IS NULL ORDER BY pk LIMIT batchLimit FOR UPDATE SKIP LOCKED 2. Decode the JS

(
	ctx restate.ObjectContext,
	_ *hydrav1.RunExportRequest,
)

// duplicate write into a noop. Marked rows stay in the table for ops to
// re-queue (clear deleted_at) and as an audit trail of what was exported.
func (s *Service) RunExport(
	ctx restate.ObjectContext,
	_ *hydrav1.RunExportRequest,
) (*hydrav1.RunExportResponse, error) {
	logger.Info("running audit log export")
	start := time.Now()

	var totalExported int32
	for batchNum := 0; ; batchNum++ {
		result, err := restate.Run(ctx, func(rc restate.RunContext) (batchResult, error) {
			return s.exportBatch(rc)
		}, restate.WithName(fmt.Sprintf("batch-%d", batchNum)))
		if err != nil {
			return nil, fmt.Errorf("batch %d: %w", batchNum, err)
		}

		totalExported += result.EventsExported

		if result.EventsExported < batchLimit {
			break
		}
	}

	logger.Info("audit log export complete",
		"events_exported", totalExported,
		"elapsed", time.Since(start),
	)

	if _, err := restate.Run(ctx, func(rc restate.RunContext) (restate.Void, error) {
		return restate.Void{}, s.heartbeat.Ping(rc)
	}, restate.WithName("send heartbeat")); err != nil {
		return nil, fmt.Errorf("send heartbeat: %w", err)
	}

	return &hydrav1.RunExportResponse{
		EventsExported: totalExported,
	}, nil
}

// exportBatch reads one batch of outbox rows, writes them to ClickHouse,
// then marks them deleted. The whole batch runs inside a single MySQL