package run

import (
	"context"
	"crypto/rand"
	"database/sql"
	"errors"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
	"github.com/tidbcloud/AgentTeam/internal/workflow/runartifact"
)

// ArtifactPage is the bounded read contract for an immutable Run artifact.
// Bytes is always at most runartifact.MaxPageBytes and NextCursor is bound to
// every identity, path, digest, and page frontier.
type ArtifactPage struct {
	Reference  runartifact.Ref
	Section    string
	Path       []string
	Bytes      []byte
	Offset     int
	TotalBytes int
	NextCursor string
}

type artifactRow struct {
	ref          runartifact.Ref
	content      []byte
	cursorSecret []byte
	status       string
	expiresAt    sql.NullTime
}

const (
	workflowArtifactRetention = 7 * 24 * time.Hour
	workflowArtifactSweepMax  = 500
)

type artifactMaintenanceKey struct {
	workspaceID string
	runID       string
	artifactID  string
	expiresAt   sql.NullTime
}

func (store *TxStore) persistArtifacts(ctx context.Context, drafts []runartifact.Draft, now time.Time, run Run) error {
	if len(drafts) == 0 {
		return nil
	}
	unique := make([]runartifact.Draft, 0, len(drafts))
	seen := make(map[string]struct{}, len(drafts))
	for _, draft := range drafts {
		if _, exists := seen[draft.Ref.ArtifactID]; exists {
			continue
		}
		seen[draft.Ref.ArtifactID] = struct{}{}
		if draft.Ref.WorkspaceID != run.WorkspaceID || draft.Ref.RunID != run.ID {
			return storeerr.BadRequest("workflow artifact belongs to a different Run")
		}
		if err := draft.Ref.Validate(); err != nil {
			return storeerr.BadRequest(err.Error())
		}
		if int64(len(draft.Content)) != draft.Ref.CanonicalByteLength || runartifact.Digest(draft.Content) != draft.Ref.Digest {
			return storeerr.BadRequest("workflow artifact content does not match its immutable digest")
		}
		unique = append(unique, draft)
	}
	if len(unique) == 0 {
		return nil
	}
	metadataSQL, metadataArgs, err := buildArtifactMetadataInsert(unique, now)
	if err != nil {
		return err
	}
	if _, err := store.tx.Exec(ctx, metadataSQL, metadataArgs...); err != nil {
		return fmt.Errorf("persist workflow artifact metadata: %w", err)
	}
	blobSQL, blobArgs := buildArtifactBlobInsert(unique)
	if _, err := store.tx.Exec(ctx, blobSQL, blobArgs...); err != nil {
		return fmt.Errorf("persist workflow artifact blobs: %w", err)
	}
	rows, err := store.tx.Query(ctx, artifactSelectSQL()+artifactIDsWhere(unique), artifactIDsArgs(unique, run.WorkspaceID, run.ID)...)
	if err != nil {
		return fmt.Errorf("load workflow artifacts after insert: %w", err)
	}
	loaded := make(map[string]artifactRow, len(unique))
	for rows.Next() {
		row, scanErr := scanArtifact(rows)
		if scanErr != nil {
			rows.Close()
			return scanErr
		}
		loaded[row.ref.ArtifactID] = row
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return fmt.Errorf("load workflow artifacts after insert: %w", err)
	}
	for _, draft := range unique {
		row, exists := loaded[draft.Ref.ArtifactID]
		if !exists || row.ref != draft.Ref || !bytesEqual(row.content, draft.Content) {
			return storeerr.Conflict("workflow artifact replay has conflicting immutable content")
		}
	}
	return nil
}

func buildArtifactMetadataInsert(drafts []runartifact.Draft, now time.Time) (string, []any, error) {
	values := make([]string, 0, len(drafts))
	args := make([]any, 0, len(drafts)*11)
	for index, draft := range drafts {
		secret := make([]byte, 32)
		if _, err := rand.Read(secret); err != nil {
			return "", nil, fmt.Errorf("generate workflow artifact cursor secret: %w", err)
		}
		base := index*11 + 1
		values = append(values, fmt.Sprintf("($%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,$%d,'finalized',$%d,$%d)", base, base+1, base+2, base+3, base+4, base+5, base+6, base+7, base+8, base+9, base+10, base+10))
		args = append(args, draft.Ref.ArtifactID, draft.Ref.WorkspaceID, draft.Ref.RunID, draft.Ref.ProducerNodeID, draft.Ref.ProducerActivationID, draft.Ref.Kind, draft.Ref.Version, draft.Ref.CanonicalByteLength, draft.Ref.Digest, secret, now)
	}
	return `INSERT INTO workflow_run_artifacts (
 id,workspace_id,run_id,producer_node_id,producer_activation_id,
 kind,version,canonical_byte_length,digest,cursor_secret,
 status,created_at,updated_at) VALUES ` + strings.Join(values, ",") + ` ON DUPLICATE KEY UPDATE updated_at=VALUES(updated_at)`, args, nil
}

func buildArtifactBlobInsert(drafts []runartifact.Draft) (string, []any) {
	values := make([]string, 0, len(drafts))
	args := make([]any, 0, len(drafts)*3)
	for index, draft := range drafts {
		base := index*3 + 1
		values = append(values, fmt.Sprintf("($%d,$%d,$%d)", base, base+1, base+2))
		args = append(args, draft.Ref.WorkspaceID, draft.Ref.ArtifactID, draft.Content)
	}
	return `INSERT INTO workflow_run_artifact_blobs (workspace_id,artifact_id,content) VALUES ` + strings.Join(values, ",") + ` ON DUPLICATE KEY UPDATE artifact_id=VALUES(artifact_id)`, args
}

func artifactIDsWhere(drafts []runartifact.Draft) string {
	placeholders := make([]string, len(drafts))
	for index := range drafts {
		placeholders[index] = fmt.Sprintf("$%d", index+3)
	}
	return ` WHERE a.workspace_id=$1 AND a.run_id=$2 AND a.id IN (` + strings.Join(placeholders, ",") + `) FOR UPDATE`
}

func artifactIDsArgs(drafts []runartifact.Draft, workspaceID, runID string) []any {
	args := make([]any, 0, len(drafts)+2)
	args = append(args, workspaceID, runID)
	for _, draft := range drafts {
		args = append(args, draft.Ref.ArtifactID)
	}
	return args
}

func (store *TxStore) loadArtifactForUpdate(ctx context.Context, ref runartifact.Ref) (artifactRow, error) {
	if err := store.Validate(); err != nil {
		return artifactRow{}, err
	}
	return scanArtifact(store.tx.QueryRow(ctx, artifactSelectSQL()+` WHERE a.workspace_id=$1 AND a.run_id=$2 AND a.id=$3 AND a.producer_node_id=$4 AND a.producer_activation_id=$5 AND a.kind=$6 AND a.version=$7 AND a.canonical_byte_length=$8 AND a.digest=$9 FOR UPDATE`,
		ref.WorkspaceID, ref.RunID, ref.ArtifactID, ref.ProducerNodeID, ref.ProducerActivationID, ref.Kind, ref.Version, ref.CanonicalByteLength, ref.Digest))
}

func (store *Store) loadArtifact(ctx context.Context, ref runartifact.Ref) (artifactRow, error) {
	if err := store.Validate(); err != nil {
		return artifactRow{}, err
	}
	if err := ref.Validate(); err != nil {
		return artifactRow{}, storeerr.BadRequest(err.Error())
	}
	return scanArtifact(store.db.QueryRow(ctx, artifactSelectSQL()+` WHERE a.workspace_id=$1 AND a.run_id=$2 AND a.id=$3 AND a.producer_node_id=$4 AND a.producer_activation_id=$5 AND a.kind=$6 AND a.version=$7 AND a.canonical_byte_length=$8 AND a.digest=$9`,
		ref.WorkspaceID, ref.RunID, ref.ArtifactID, ref.ProducerNodeID, ref.ProducerActivationID, ref.Kind, ref.Version, ref.CanonicalByteLength, ref.Digest))
}

func artifactSelectSQL() string {
	return `SELECT a.id,a.workspace_id,a.run_id,a.producer_node_id,a.producer_activation_id,
 a.kind,a.version,a.canonical_byte_length,a.digest,b.content,a.cursor_secret,a.status,a.expires_at
 FROM workflow_run_artifacts a
 JOIN workflow_run_artifact_blobs b ON b.workspace_id=a.workspace_id AND b.artifact_id=a.id`
}

func scanArtifact(row interface{ Scan(...any) error }) (artifactRow, error) {
	var result artifactRow
	var ref runartifact.Ref
	if err := row.Scan(&ref.ArtifactID, &ref.WorkspaceID, &ref.RunID, &ref.ProducerNodeID, &ref.ProducerActivationID,
		&ref.Kind, &ref.Version, &ref.CanonicalByteLength, &ref.Digest,
		&result.content, &result.cursorSecret, &result.status, &result.expiresAt); err != nil {
		if errors.Is(err, sql.ErrNoRows) || coredb.IsNotFound(err) {
			return artifactRow{}, storeerr.ErrNotFound
		}
		return artifactRow{}, fmt.Errorf("load workflow artifact: %w", err)
	}
	result.ref = ref
	if err := ref.Validate(); err != nil {
		return artifactRow{}, storeerr.ErrConflict
	}
	if int64(len(result.content)) != ref.CanonicalByteLength || runartifact.Digest(result.content) != ref.Digest || result.status != "finalized" {
		return artifactRow{}, storeerr.ErrConflict
	}
	if result.expiresAt.Valid && !result.expiresAt.Time.After(time.Now().UTC()) {
		return artifactRow{}, storeerr.ErrConflict
	}
	return result, nil
}

func bytesEqual(left, right []byte) bool {
	if len(left) != len(right) {
		return false
	}
	for index := range left {
		if left[index] != right[index] {
			return false
		}
	}
	return true
}

// ResolveArtifactContent authorizes and verifies one exact reference before a
// trusted effect dispatcher materializes it for external delivery.
func (store *Store) ResolveArtifactContent(ctx context.Context, ref runartifact.Ref) ([]byte, error) {
	if err := ref.Validate(); err != nil {
		return nil, storeerr.BadRequest(err.Error())
	}
	var status string
	if err := store.db.QueryRow(ctx, `SELECT status FROM workflow_run_instances WHERE workspace_id=$1 AND id=$2`, ref.WorkspaceID, ref.RunID).Scan(&status); err != nil {
		if coredb.IsNotFound(err) || errors.Is(err, sql.ErrNoRows) {
			return nil, storeerr.ErrNotFound
		}
		return nil, fmt.Errorf("authorize workflow artifact content: %w", err)
	}
	if status == RunStatusCanceled {
		return nil, storeerr.ErrConflict
	}
	row, err := store.loadArtifact(ctx, ref)
	if err != nil {
		return nil, err
	}
	return append([]byte(nil), row.content...), nil
}

// MaterializeRunState expands only artifacts bound to this exact workspace/Run.
// A marker's artifact ID is never sufficient: loadArtifact matches every
// producer/version/digest field and verifies the immutable bytes.
func (store *Store) MaterializeRunState(ctx context.Context, workspaceID, runID string, stateJSON []byte) ([]byte, error) {
	workspaceID, runID = strings.TrimSpace(workspaceID), strings.TrimSpace(runID)
	if workspaceID == "" || runID == "" {
		return nil, storeerr.BadRequest("workflow artifact materialization requires workspaceId and runId")
	}
	var runStatus string
	if err := store.db.QueryRow(ctx, `SELECT status FROM workflow_run_instances WHERE workspace_id=$1 AND id=$2`, workspaceID, runID).Scan(&runStatus); err != nil {
		if coredb.IsNotFound(err) || errors.Is(err, sql.ErrNoRows) {
			return nil, storeerr.ErrNotFound
		}
		return nil, fmt.Errorf("authorize workflow artifact materialization: %w", err)
	}
	if runStatus == RunStatusCanceled {
		return nil, storeerr.ErrConflict
	}
	return runartifact.ExpandJSON(stateJSON, func(ref runartifact.Ref) ([]byte, error) {
		if ref.WorkspaceID != workspaceID || ref.RunID != runID {
			return nil, storeerr.ErrConflict
		}
		row, err := store.loadArtifact(ctx, ref)
		if err != nil {
			return nil, err
		}
		return append([]byte(nil), row.content...), nil
	})
}

// ReadArtifactPage authorizes the complete reference against the Run and
// artifact row before reading bytes. The row is locked for the short read so a
// future retention sweep cannot delete an artifact between authorization and
// page extraction.
func (store *Store) ReadArtifactPage(ctx context.Context, ref runartifact.Ref, section string, path []string, cursor string, limit int) (ArtifactPage, error) {
	if ref.Version != runartifact.Version {
		return ArtifactPage{}, runartifact.ErrStaleReference
	}
	if err := ref.Validate(); err != nil {
		return ArtifactPage{}, err
	}
	if limit == 0 {
		limit = runartifact.DefaultPageBytes
	}
	if limit < 1 || limit > runartifact.MaxPageBytes {
		return ArtifactPage{}, fmt.Errorf("%w: workflow artifact page limit is invalid", runartifact.ErrInvalidRead)
	}
	var page ArtifactPage
	err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		var runStatus string
		if err := tx.QueryRow(ctx, `SELECT status FROM workflow_run_instances WHERE workspace_id=$1 AND id=$2 FOR UPDATE`, ref.WorkspaceID, ref.RunID).Scan(&runStatus); err != nil {
			if coredb.IsNotFound(err) || errors.Is(err, sql.ErrNoRows) {
				return storeerr.ErrNotFound
			}
			return fmt.Errorf("authorize workflow artifact page: %w", err)
		}
		if runStatus == RunStatusCanceled {
			return storeerr.ErrConflict
		}
		row, err := (&TxStore{tx: tx}).loadArtifactForUpdate(ctx, ref)
		if err != nil {
			if errors.Is(err, storeerr.ErrConflict) {
				return runartifact.ErrStaleReference
			}
			if errors.Is(err, storeerr.ErrNotFound) {
				var storedKind, storedVersion, storedDigest string
				metadataErr := tx.QueryRow(ctx, `SELECT kind, version, digest FROM workflow_run_artifacts WHERE workspace_id=$1 AND run_id=$2 AND id=$3`, ref.WorkspaceID, ref.RunID, ref.ArtifactID).Scan(&storedKind, &storedVersion, &storedDigest)
				if metadataErr == nil && (storedKind != ref.Kind || storedVersion != ref.Version || storedDigest != ref.Digest) {
					return runartifact.ErrStaleReference
				}
				return runartifact.ErrUnauthorizedReference
			}
			return err
		}
		selected, err := runartifact.SelectPath(row.content, path)
		if err != nil {
			return err
		}
		offset := 0
		if cursor != "" {
			decoded, err := runartifact.DecodeCursor(cursor, ref, section, path, limit, row.cursorSecret)
			if err != nil {
				return err
			}
			offset = decoded.Offset
		}
		if offset > len(selected) {
			return fmt.Errorf("%w: workflow artifact cursor is past the content frontier", runartifact.ErrInvalidCursor)
		}
		end := offset + limit
		if end > len(selected) {
			end = len(selected)
		}
		page = ArtifactPage{Reference: ref, Section: section, Path: append([]string(nil), path...), Bytes: append([]byte(nil), selected[offset:end]...), Offset: offset, TotalBytes: len(selected)}
		if end < len(selected) {
			page.NextCursor, err = runartifact.EncodeCursor(ref, section, path, end, limit, row.cursorSecret)
			if err != nil {
				return err
			}
		}
		return nil
	})
	return page, err
}

// RunArtifactMaintenance processes at most limit canonical artifact keys in
// one transaction. It locks each parent Run before its artifact, assigns
// terminal retention only for those keys, deletes only already-expired keys,
// and spends any remaining budget on orphan blobs. Page authorization uses the
// same lock order, so live reads/callbacks cannot be detached by cleanup.
func (store *Store) RunArtifactMaintenance(ctx context.Context, now time.Time, limit int) (ArtifactMaintenanceResult, error) {
	if err := store.Validate(); err != nil {
		return ArtifactMaintenanceResult{}, err
	}
	if limit <= 0 || limit > workflowArtifactSweepMax {
		return ArtifactMaintenanceResult{}, storeerr.BadRequest("workflow artifact maintenance limit is invalid")
	}
	now = now.UTC()
	var result ArtifactMaintenanceResult
	err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		keys, err := listArtifactMaintenanceKeys(ctx, tx, now, limit)
		if err != nil {
			return err
		}
		if len(keys) > 0 {
			if err := lockArtifactMaintenanceRuns(ctx, tx, keys); err != nil {
				return err
			}
			locked, err := lockArtifactMaintenanceArtifacts(ctx, tx, keys)
			if err != nil {
				return err
			}
			assign := make([]artifactMaintenanceKey, 0, len(locked))
			deleteKeys := make([]artifactMaintenanceKey, 0, len(locked))
			for _, key := range locked {
				if !key.expiresAt.Valid {
					assign = append(assign, key)
				} else if !key.expiresAt.Time.After(now) {
					deleteKeys = append(deleteKeys, key)
				}
			}
			if len(assign) > 0 {
				assigned, err := assignArtifactRetention(ctx, tx, assign, now)
				if err != nil {
					return err
				}
				result.AssignedExpiry = assigned
			}
			if len(deleteKeys) > 0 {
				deleted, err := deleteExpiredArtifactKeys(ctx, tx, deleteKeys, now)
				if err != nil {
					return err
				}
				result.Deleted = deleted
			}
		}
		remaining := limit - len(keys)
		if remaining > 0 {
			orphanKeys, err := listOrphanArtifactBlobs(ctx, tx, remaining)
			if err != nil {
				return err
			}
			if len(orphanKeys) > 0 {
				query, args := artifactBlobDeleteQuery(orphanKeys)
				orphaned, err := tx.Exec(ctx, query, args...)
				if err != nil {
					return fmt.Errorf("cleanup orphan workflow artifact blobs: %w", err)
				}
				result.OrphanedBlobs = orphaned.RowsAffected()
			}
		}
		return nil
	})
	return result, err
}

func listArtifactMaintenanceKeys(ctx context.Context, tx coredb.Tx, now time.Time, limit int) ([]artifactMaintenanceKey, error) {
	rows, err := tx.Query(ctx, `
SELECT a.workspace_id,a.run_id,a.id,a.expires_at
FROM workflow_run_artifacts a
JOIN workflow_run_instances r ON r.workspace_id=a.workspace_id AND r.id=a.run_id
WHERE a.status='finalized' AND (a.expires_at IS NULL OR a.expires_at <= $1)
  AND r.status IN ('succeeded','failed','canceled')
  AND NOT EXISTS (SELECT 1 FROM workflow_run_activations x WHERE x.workspace_id=a.workspace_id AND x.run_id=a.run_id AND x.status IN ('ready','waiting'))
  AND NOT EXISTS (SELECT 1 FROM workflow_run_effects e WHERE e.workspace_id=a.workspace_id AND e.run_id=a.run_id AND e.status IN ('pending','dispatching','delivered','awaiting','waiting_for_responder'))
ORDER BY CASE WHEN a.expires_at IS NULL THEN 1 ELSE 0 END,a.expires_at,a.workspace_id,a.id LIMIT $2`, now, limit)
	if err != nil {
		return nil, fmt.Errorf("list workflow artifact maintenance keys: %w", err)
	}
	defer rows.Close()
	keys := make([]artifactMaintenanceKey, 0, limit)
	for rows.Next() {
		var key artifactMaintenanceKey
		if err := rows.Scan(&key.workspaceID, &key.runID, &key.artifactID, &key.expiresAt); err != nil {
			return nil, fmt.Errorf("scan workflow artifact maintenance key: %w", err)
		}
		keys = append(keys, key)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("scan workflow artifact maintenance keys: %w", err)
	}
	return keys, nil
}

func lockArtifactMaintenanceRuns(ctx context.Context, tx coredb.Tx, keys []artifactMaintenanceKey) error {
	clauses, args := pairPredicates("r.workspace_id", "r.id", keys, func(key artifactMaintenanceKey) (string, string) {
		return key.workspaceID, key.runID
	})
	rows, err := tx.Query(ctx, `SELECT r.workspace_id,r.id FROM workflow_run_instances r WHERE `+clauses+` ORDER BY r.workspace_id,r.id FOR UPDATE`, args...)
	if err != nil {
		return fmt.Errorf("lock workflow artifact Runs: %w", err)
	}
	defer rows.Close()
	for rows.Next() {
		var workspaceID, runID string
		if err := rows.Scan(&workspaceID, &runID); err != nil {
			return fmt.Errorf("scan locked workflow artifact Run: %w", err)
		}
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("scan locked workflow artifact Runs: %w", err)
	}
	return nil
}

func lockArtifactMaintenanceArtifacts(ctx context.Context, tx coredb.Tx, keys []artifactMaintenanceKey) ([]artifactMaintenanceKey, error) {
	clauses, args := pairPredicates("a.workspace_id", "a.id", keys, func(key artifactMaintenanceKey) (string, string) {
		return key.workspaceID, key.artifactID
	})
	rows, err := tx.Query(ctx, `SELECT a.workspace_id,a.run_id,a.id,a.expires_at FROM workflow_run_artifacts a WHERE a.status='finalized' AND `+clauses+` ORDER BY a.workspace_id,a.id FOR UPDATE`, args...)
	if err != nil {
		return nil, fmt.Errorf("lock workflow artifacts: %w", err)
	}
	defer rows.Close()
	locked := make([]artifactMaintenanceKey, 0, len(keys))
	for rows.Next() {
		var key artifactMaintenanceKey
		if err := rows.Scan(&key.workspaceID, &key.runID, &key.artifactID, &key.expiresAt); err != nil {
			return nil, fmt.Errorf("scan locked workflow artifact: %w", err)
		}
		locked = append(locked, key)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("scan locked workflow artifacts: %w", err)
	}
	return locked, nil
}

func assignArtifactRetention(ctx context.Context, tx coredb.Tx, keys []artifactMaintenanceKey, now time.Time) (int64, error) {
	clauses, args := pairPredicates("a.workspace_id", "a.id", keys, func(key artifactMaintenanceKey) (string, string) {
		return key.workspaceID, key.artifactID
	})
	args = append(args, now.Add(workflowArtifactRetention), now)
	result, err := tx.Exec(ctx, `UPDATE workflow_run_artifacts a
JOIN workflow_run_instances r ON r.workspace_id=a.workspace_id AND r.id=a.run_id
SET a.expires_at=$`+strconv.Itoa(len(args)-1)+`,a.updated_at=$`+strconv.Itoa(len(args))+`
WHERE `+clauses+` AND a.expires_at IS NULL AND r.status IN ('succeeded','failed','canceled')
  AND NOT EXISTS (SELECT 1 FROM workflow_run_activations x WHERE x.workspace_id=a.workspace_id AND x.run_id=a.run_id AND x.status IN ('ready','waiting'))
  AND NOT EXISTS (SELECT 1 FROM workflow_run_effects e WHERE e.workspace_id=a.workspace_id AND e.run_id=a.run_id AND e.status IN ('pending','dispatching','delivered','awaiting','waiting_for_responder'))`, args...)
	if err != nil {
		return 0, fmt.Errorf("assign terminal workflow artifact retention: %w", err)
	}
	return result.RowsAffected(), nil
}

func deleteExpiredArtifactKeys(ctx context.Context, tx coredb.Tx, keys []artifactMaintenanceKey, now time.Time) (int64, error) {
	clauses, args := pairPredicates("a.workspace_id", "a.id", keys, func(key artifactMaintenanceKey) (string, string) {
		return key.workspaceID, key.artifactID
	})
	args = append(args, now)
	result, err := tx.Exec(ctx, `DELETE FROM workflow_run_artifacts a
WHERE `+clauses+` AND a.status='finalized' AND a.expires_at IS NOT NULL AND a.expires_at <= $`+strconv.Itoa(len(args))+`
  AND EXISTS (SELECT 1 FROM workflow_run_instances r WHERE r.workspace_id=a.workspace_id AND r.id=a.run_id AND r.status IN ('succeeded','failed','canceled'))
  AND NOT EXISTS (SELECT 1 FROM workflow_run_activations x WHERE x.workspace_id=a.workspace_id AND x.run_id=a.run_id AND x.status IN ('ready','waiting'))
  AND NOT EXISTS (SELECT 1 FROM workflow_run_effects e WHERE e.workspace_id=a.workspace_id AND e.run_id=a.run_id AND e.status IN ('pending','dispatching','delivered','awaiting','waiting_for_responder'))`, args...)
	if err != nil {
		return 0, fmt.Errorf("delete expired workflow artifacts: %w", err)
	}
	return result.RowsAffected(), nil
}

func pairPredicates(left, right string, keys []artifactMaintenanceKey, values func(artifactMaintenanceKey) (string, string)) (string, []any) {
	clauses := make([]string, 0, len(keys))
	args := make([]any, 0, len(keys)*2)
	for index, key := range keys {
		leftValue, rightValue := values(key)
		base := index*2 + 1
		clauses = append(clauses, fmt.Sprintf("(%s=$%d AND %s=$%d)", left, base, right, base+1))
		args = append(args, leftValue, rightValue)
	}
	return strings.Join(clauses, " OR "), args
}

func artifactKeyDeleteQuery(keys [][2]string) (string, []any) {
	clauses := make([]string, 0, len(keys))
	args := make([]any, 0, len(keys)*2)
	for index, key := range keys {
		base := index*2 + 1
		clauses = append(clauses, fmt.Sprintf("(workspace_id=$%d AND id=$%d)", base, base+1))
		args = append(args, key[0], key[1])
	}
	return `DELETE FROM workflow_run_artifacts WHERE ` + strings.Join(clauses, " OR "), args
}

// SweepExpiredArtifacts is the compatibility entry point used by explicit
// maintenance callers. It shares the production lifecycle transaction rather
// than deleting an expired child solely because expires_at was manually set.
func (store *Store) SweepExpiredArtifacts(ctx context.Context, now time.Time, limit int) (int64, error) {
	result, err := store.RunArtifactMaintenance(ctx, now, limit)
	return result.Deleted, err
}

// CleanupOrphanArtifactBlobs performs the same bounded orphan-only pass used
// by RunArtifactMaintenance, retaining the explicit repair API for operators.
func (store *Store) CleanupOrphanArtifactBlobs(ctx context.Context, limit int) (int64, error) {
	if err := store.Validate(); err != nil {
		return 0, err
	}
	if limit <= 0 || limit > workflowArtifactSweepMax {
		return 0, storeerr.BadRequest("workflow artifact orphan cleanup limit is invalid")
	}
	var deleted int64
	err := coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		keys, err := listOrphanArtifactBlobs(ctx, tx, limit)
		if err != nil {
			return err
		}
		if len(keys) == 0 {
			return nil
		}
		query, args := artifactBlobDeleteQuery(keys)
		result, err := tx.Exec(ctx, query, args...)
		if err != nil {
			return fmt.Errorf("cleanup orphan workflow artifact blobs: %w", err)
		}
		deleted = result.RowsAffected()
		return nil
	})
	return deleted, err
}

func listOrphanArtifactBlobs(ctx context.Context, tx coredb.Tx, limit int) ([][2]string, error) {
	rows, err := tx.Query(ctx, `
SELECT b.workspace_id,b.artifact_id
FROM workflow_run_artifact_blobs b
LEFT JOIN workflow_run_artifacts a ON a.workspace_id=b.workspace_id AND a.id=b.artifact_id
WHERE a.id IS NULL ORDER BY b.workspace_id,b.artifact_id LIMIT $1`, limit)
	if err != nil {
		return nil, fmt.Errorf("list orphan workflow artifact blobs: %w", err)
	}
	defer rows.Close()
	keys := make([][2]string, 0, limit)
	for rows.Next() {
		var key [2]string
		if err := rows.Scan(&key[0], &key[1]); err != nil {
			return nil, fmt.Errorf("scan orphan workflow artifact blob: %w", err)
		}
		keys = append(keys, key)
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("scan orphan workflow artifact blobs: %w", err)
	}
	return keys, nil
}

func artifactBlobDeleteQuery(keys [][2]string) (string, []any) {
	clauses := make([]string, 0, len(keys))
	args := make([]any, 0, len(keys)*2)
	for index, key := range keys {
		base := index*2 + 1
		clauses = append(clauses, fmt.Sprintf("(workspace_id=$%d AND artifact_id=$%d)", base, base+1))
		args = append(args, key[0], key[1])
	}
	return `DELETE FROM workflow_run_artifact_blobs WHERE ` + strings.Join(clauses, " OR "), args
}