Add log export for audit logs and SCIM events

Route audit-log and SCIM-event exports through export_jobs with typed
arguments, an iam BuildAndUploadExport/SendExportEmail implementation,
and a concurrent export-job worker with stale recovery. Stream JSONL via
page.WalkAll into S3, and expose the request flow on console, connect,
MCP, and CLI.

Co-authored-by: Bryan Frimin <bryan@getprobo.com>
Signed-off-by: Sacha Al Himdani <sacha@probo.com>

pkg/coredata/export_job.go

@@ -63,6 +63,11 @@ type (
 	FrameworkExportArguments struct {
 		FrameworkID gid.GID `json:"framework_id"`
 	}
+
+	LogExportArguments struct {
+		FromTime time.Time `json:"from_time"`
+		ToTime   time.Time `json:"to_time"`
+	}
 )
 
 var (
@@ -164,7 +169,6 @@ SET
 	status = @status,
 	error = @error,
 	file_id = @file_id,
-	started_at = @started_at,
 	completed_at = @completed_at
 WHERE
 	%s
@@ -175,14 +179,110 @@ WHERE
 		"status":       ej.Status,
 		"error":        ej.Error,
 		"file_id":      ej.FileID,
-		"started_at":   ej.StartedAt,
 		"completed_at": ej.CompletedAt,
 		"id":           ej.ID,
 	}
 	maps.Copy(args, scope.SQLArguments())
-	_, err := conn.Exec(ctx, q, args)
 
-	return err
+	result, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot update export job: %w", err)
+	}
+
+	if result.RowsAffected() == 0 {
+		return ErrResourceNotFound
+	}
+
+	return nil
+}
+
+// UpdateIfStatus updates the export job only when its current status matches
+// expected. Used to finalize a claim without clobbering a job that was
+// reclaimed and reassigned to another worker.
+func (ej *ExportJob) UpdateIfStatus(
+	ctx context.Context,
+	conn pg.Tx,
+	scope Scoper,
+	expected ExportJobStatus,
+) error {
+	q := `
+UPDATE
+	export_jobs
+SET
+	status = @status,
+	error = @error,
+	file_id = @file_id,
+	completed_at = @completed_at
+WHERE
+	%s
+	AND id = @id
+	AND status = @expected_status
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+	args := pgx.StrictNamedArgs{
+		"status":          ej.Status,
+		"error":           ej.Error,
+		"file_id":         ej.FileID,
+		"completed_at":    ej.CompletedAt,
+		"id":              ej.ID,
+		"expected_status": expected,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	result, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot update export job: %w", err)
+	}
+
+	if result.RowsAffected() == 0 {
+		return ErrResourceNotFound
+	}
+
+	return nil
+}
+
+// MarkProcessing claims the job for processing and starts its lease clock.
+func (ej *ExportJob) MarkProcessing(
+	ctx context.Context,
+	conn pg.Tx,
+	scope Scoper,
+) error {
+	now := time.Now()
+	ej.Status = ExportJobStatusProcessing
+	ej.StartedAt = &now
+
+	q := `
+UPDATE
+	export_jobs
+SET
+	status = @status,
+	started_at = @started_at,
+	error = NULL,
+	completed_at = NULL
+WHERE
+	%s
+	AND id = @id
+	AND status = @pending_status
+`
+	q = fmt.Sprintf(q, scope.SQLFragment())
+	args := pgx.StrictNamedArgs{
+		"status":         ej.Status,
+		"started_at":     ej.StartedAt,
+		"id":             ej.ID,
+		"pending_status": ExportJobStatusPending,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	result, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot mark export job as processing: %w", err)
+	}
+
+	if result.RowsAffected() == 0 {
+		return ErrResourceNotFound
+	}
+
+	return nil
 }
 
 func (ej *ExportJob) LoadByID(
@@ -323,3 +423,85 @@ func (ej *ExportJob) GetFrameworkID() (gid.GID, error) {
 
 	return args.FrameworkID, nil
 }
+
+func (ej *ExportJob) GetLogExportArguments() (*LogExportArguments, error) {
+	switch ej.Type {
+	case ExportJobTypeAuditLog, ExportJobTypeSCIMEvent:
+	default:
+		return nil, fmt.Errorf("export job is not a log export")
+	}
+
+	var args LogExportArguments
+	if err := json.Unmarshal(ej.Arguments, &args); err != nil {
+		return nil, fmt.Errorf("cannot unmarshal log export arguments: %w", err)
+	}
+
+	return &args, nil
+}
+
+func ResetStaleExportJobs(
+	ctx context.Context,
+	conn pg.Querier,
+	staleAfter time.Duration,
+) error {
+	q := `
+UPDATE export_jobs
+SET
+	status = @pending_status,
+	started_at = NULL
+WHERE
+	status = @processing_status
+	AND started_at < @stale_threshold
+`
+
+	args := pgx.StrictNamedArgs{
+		"pending_status":    ExportJobStatusPending,
+		"processing_status": ExportJobStatusProcessing,
+		"stale_threshold":   time.Now().Add(-staleAfter),
+	}
+
+	_, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot reset stale export jobs: %w", err)
+	}
+
+	return nil
+}
+
+// TouchExportJobLease renews the processing lease so long-running exports
+// are not requeued by ResetStaleExportJobs while still actively working.
+func TouchExportJobLease(
+	ctx context.Context,
+	conn pg.Querier,
+	scope Scoper,
+	id gid.GID,
+) error {
+	q := `
+UPDATE export_jobs
+SET
+	started_at = @started_at
+WHERE
+	%s
+	AND id = @id
+	AND status = @processing_status
+`
+
+	q = fmt.Sprintf(q, scope.SQLFragment())
+	args := pgx.StrictNamedArgs{
+		"id":                id,
+		"started_at":        time.Now(),
+		"processing_status": ExportJobStatusProcessing,
+	}
+	maps.Copy(args, scope.SQLArguments())
+
+	result, err := conn.Exec(ctx, q, args)
+	if err != nil {
+		return fmt.Errorf("cannot touch export job lease: %w", err)
+	}
+
+	if result.RowsAffected() == 0 {
+		return ErrResourceNotFound
+	}
+
+	return nil
+}