package requests import ( "context" "errors" "fmt" "github.com/jackc/pgx/v5" ) const targetColumns = `t.id, t.request_id, t.integration_id, t.integration_kind, COALESCE(ri.name, ''), t.quality, t.is_anime, t.external_id, t.external_status, t.status, t.last_error, t.created_at, t.updated_at` // aggregateStatus derives a request's status/outcome from its targets. func aggregateStatus(targets []Target) (Status, Outcome) { if len(targets) == 0 { return StatusApproved, OutcomeActive } failed, completed := 0, 0 anyDownloading, anyQueued := false, false for _, t := range targets { switch t.Status { case StatusFailed: failed++ case StatusCompleted: completed++ case StatusDownloading: anyDownloading = true case StatusQueued: anyQueued = true } } if completed == len(targets) { return StatusCompleted, OutcomeActive } // Active targets keep the request active even with a failed sibling so the // in-flight targets can finish (partial failure stays active). if anyDownloading { return StatusDownloading, OutcomeActive } if anyQueued { return StatusQueued, OutcomeActive } // No active targets remain and at least one failed (all-failed, or a mix of // completed + failed) -> surface as failed so Retry can re-submit the failed // target while leaving completed ones untouched. if failed > 0 { return StatusQueued, OutcomeFailed } return StatusCompleted, OutcomeActive } func scanTarget(row requestScanner) (Target, error) { var t Target var integrationID *string if err := row.Scan(&t.ID, &t.RequestID, &integrationID, &t.IntegrationKind, &t.InstanceName, &t.Quality, &t.IsAnime, &t.ExternalID, &t.ExternalStatus, &t.Status, &t.LastError, &t.CreatedAt, &t.UpdatedAt); err != nil { return Target{}, err } if integrationID != nil { t.IntegrationID = *integrationID } return t, nil } func (r *Repository) ListTargets(ctx context.Context, requestID string) ([]Target, error) { rows, err := r.pool.Query(ctx, `SELECT `+targetColumns+` FROM media_request_targets t LEFT JOIN request_integrations ri ON ri.id = t.integration_id WHERE t.request_id = $1 ORDER BY t.quality`, requestID) if err != nil { return nil, fmt.Errorf("list targets: %w", err) } defer rows.Close() var out []Target for rows.Next() { t, err := scanTarget(rows) if err != nil { return nil, err } out = append(out, t) } return out, rows.Err() } func (r *Repository) CreateTarget(ctx context.Context, t Target) (Target, error) { var integrationID any if t.IntegrationID != "" { integrationID = t.IntegrationID } row := r.pool.QueryRow(ctx, ` INSERT INTO media_request_targets (request_id, integration_id, integration_kind, quality, is_anime, external_id, external_status, status, last_error, updated_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9, now()) RETURNING id`, t.RequestID, integrationID, t.IntegrationKind, t.Quality, t.IsAnime, t.ExternalID, t.ExternalStatus, t.Status, t.LastError) if err := row.Scan(&t.ID); err != nil { return Target{}, fmt.Errorf("create target: %w", err) } return t, nil } func (r *Repository) DeleteTarget(ctx context.Context, id int64) error { tag, err := r.pool.Exec(ctx, `DELETE FROM media_request_targets WHERE id = $1`, id) if err != nil { return fmt.Errorf("delete target: %w", err) } if tag.RowsAffected() == 0 { return ErrNotFound } return nil } // UpdateTargetStatus updates one target and recomputes the parent request's // aggregate status/outcome, all in one transaction. func (r *Repository) UpdateTargetStatus(ctx context.Context, targetID int64, status Status, externalID, externalStatus, lastErr string, actor Viewer) (*Request, error) { tx, err := r.pool.Begin(ctx) if err != nil { return nil, fmt.Errorf("begin target update: %w", err) } defer tx.Rollback(ctx) var requestID string if err := tx.QueryRow(ctx, ` UPDATE media_request_targets SET status=$2, external_id = CASE WHEN $3 = '' THEN external_id ELSE $3 END, external_status = CASE WHEN $4 = '' THEN external_status ELSE $4 END, last_error=$5, updated_at=now() WHERE id=$1 RETURNING request_id`, targetID, status, externalID, externalStatus, lastErr).Scan(&requestID); err != nil { if errors.Is(err, pgx.ErrNoRows) { return nil, ErrNotFound } return nil, fmt.Errorf("update target: %w", err) } req, err := r.recomputeAggregate(ctx, tx, requestID, actor) if err != nil { return nil, err } if err := tx.Commit(ctx); err != nil { return nil, fmt.Errorf("commit target update: %w", err) } return req, nil } func (r *Repository) recomputeAggregate(ctx context.Context, exec requestExecutor, requestID string, actor Viewer) (*Request, error) { rows, err := exec.Query(ctx, `SELECT status FROM media_request_targets WHERE request_id = $1`, requestID) if err != nil { return nil, fmt.Errorf("load target statuses: %w", err) } var targets []Target for rows.Next() { var t Target if err := rows.Scan(&t.Status); err != nil { rows.Close() return nil, err } targets = append(targets, t) } rows.Close() if err := rows.Err(); err != nil { return nil, err } status, outcome := aggregateStatus(targets) var lastErr string for _, t := range targets { if t.Status == StatusFailed { lastErr = "one or more fulfillment targets failed" break } } req, err := scanRequest(exec.QueryRow(ctx, ` UPDATE media_requests SET status=$2, outcome=$3, last_error = CASE WHEN $3 = 'failed' THEN $4 ELSE '' END, completed_at = CASE WHEN $2 = 'completed' AND completed_at IS NULL THEN now() ELSE completed_at END, updated_at = now() WHERE id=$1 RETURNING `+requestColumns(), requestID, status, outcome, lastErr)) if err != nil { return nil, fmt.Errorf("recompute aggregate: %w", err) } _ = r.recordEvent(ctx, exec, requestID, "status_"+string(status), actor, string(req.ExternalStatus)) return req, nil }