package submit

import (
	"context"
	"database/sql"
	"time"
)

// PGStore is the Postgres-backed Store (image_submissions, migration 0002). It
// is the only submit component that holds database credentials and exposes only
// the narrow operations the Manager needs — no generic UPDATE escape hatch. It
// compiles here but is exercised by integration tests against a live database;
// the Manager's logic is unit-tested against the in-memory fake instead.
type PGStore struct {
	db *sql.DB
}

// NewPGStore wraps an open *sql.DB.
func NewPGStore(db *sql.DB) *PGStore { return &PGStore{db: db} }

// Compile-time proof PGStore satisfies the Store interface.
var _ Store = (*PGStore)(nil)

const submissionColumns = `id, submitted_by, display_name, context_ref, status,
	image_ref, build_id, reviewed_by, reject_reason, created_at, reviewed_at, context_sha256`

// CreateSubmission counts and inserts in one transaction under a per-submitter
// advisory lock, so two creates by the same user, on one replica or several,
// serialize and the second sees the first's row. A hashtext collision between
// two users only serializes their creates.
func (s *PGStore) CreateSubmission(ctx context.Context, sub *Submission, maxPending int) (int, error) {
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return 0, err
	}
	defer func() { _ = tx.Rollback() }()
	if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtext('submission:' || $1))`, sub.SubmittedBy); err != nil {
		return 0, err
	}
	var pending int
	if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM image_submissions
		WHERE submitted_by = $1 AND status = 'pending_review'`, sub.SubmittedBy).Scan(&pending); err != nil {
		return 0, err
	}
	if pending >= maxPending {
		return pending, ErrQuotaExceeded
	}
	const q = `INSERT INTO image_submissions
		(id, submitted_by, display_name, context_ref, status, created_at)
		VALUES ($1, $2, $3, $4, $5, $6)`
	if _, err := tx.ExecContext(ctx, q,
		sub.ID, sub.SubmittedBy, sub.DisplayName, sub.ContextRef, string(sub.Status), sub.CreatedAt); err != nil {
		return pending, err
	}
	return pending, tx.Commit()
}

func (s *PGStore) CountPendingSubmissionsBy(ctx context.Context, submittedBy string) (int, error) {
	const q = `SELECT count(*) FROM image_submissions
		WHERE submitted_by = $1 AND status = 'pending_review'`
	var n int
	err := s.db.QueryRowContext(ctx, q, submittedBy).Scan(&n)
	return n, err
}

func (s *PGStore) GetSubmission(ctx context.Context, id string) (*Submission, error) {
	const q = `SELECT ` + submissionColumns + ` FROM image_submissions WHERE id = $1`
	return scanSubmission(s.db.QueryRowContext(ctx, q, id))
}

func (s *PGStore) ListSubmissions(ctx context.Context) ([]Submission, error) {
	const q = `SELECT ` + submissionColumns + ` FROM image_submissions ORDER BY created_at DESC`
	return s.querySubmissions(ctx, q)
}

// PageSubmissions reads the scope's status counts, the filtered total and one
// page. id breaks created_at ties so a page boundary never repeats or skips a row.
func (s *PGStore) PageSubmissions(ctx context.Context, opts ListOpts) (Page, error) {
	page := Page{Counts: map[Status]int{}}
	rows, err := s.db.QueryContext(ctx, `SELECT status, count(*) FROM image_submissions
		WHERE $1::text = '' OR submitted_by = $1::text GROUP BY status`, opts.SubmittedBy)
	if err != nil {
		return Page{}, err
	}
	defer rows.Close()
	for rows.Next() {
		var st string
		var n int
		if err := rows.Scan(&st, &n); err != nil {
			return Page{}, err
		}
		page.Counts[Status(st)] = n
	}
	if err := rows.Err(); err != nil {
		return Page{}, err
	}
	const match = ` WHERE ($1::text = '' OR submitted_by = $1::text)
		AND ($2::text = '' OR status::text = $2::text)
		AND ($3::text = '' OR strpos(lower(id), lower($3::text)) > 0
			OR strpos(lower(submitted_by), lower($3::text)) > 0
			OR strpos(lower(display_name), lower($3::text)) > 0)`
	args := []any{opts.SubmittedBy, string(opts.Status), opts.Query}
	if err := s.db.QueryRowContext(ctx, `SELECT count(*) FROM image_submissions`+match, args...).Scan(&page.Total); err != nil {
		return Page{}, err
	}
	page.Submissions, err = s.querySubmissions(ctx, `SELECT `+submissionColumns+` FROM image_submissions`+match+`
		ORDER BY created_at DESC, id DESC LIMIT $4 OFFSET $5`, append(args, opts.Limit, opts.Offset)...)
	return page, err
}

// cas executes a single-statement compare-and-set — an UPDATE or DELETE whose
// WHERE clause is the predicate — and reports whether THIS call moved a row. The
// `status = 'pending_review'` guard is the actual CAS predicate in each caller's
// query and is deliberately kept INLINE — it is security-visible, so a reader
// auditing "can a non-pending row be flipped or deleted?" must see it next to the
// SET or DELETE. cas only folds the shared ExecContext + RowsAffected tail so
// the reviewers and deleters cannot drift in how they report a lost race or a
// RowsAffected error. n == 0 means a concurrent actor already won the row —
// reported as won=false (never an error), which the Manager maps to
// ErrAlreadyReviewed / ErrNotFound.
func (s *PGStore) cas(ctx context.Context, q string, args ...any) (bool, error) {
	res, err := s.db.ExecContext(ctx, q, args...)
	if err != nil {
		return false, err
	}
	n, err := res.RowsAffected()
	return n > 0, err
}

// ApproveSubmission is the approve CAS: it flips the row only while it is still
// pending_review AND still carries the digest the reviewer approved, so neither
// a concurrent reviewer nor a re-upload after the review can let it win.
func (s *PGStore) ApproveSubmission(ctx context.Context, id, reviewedBy, imageRef, digest string, at time.Time) (bool, error) {
	const q = `UPDATE image_submissions
		SET status = 'approved', image_ref = $2, reviewed_by = $3, reviewed_at = $4
		WHERE id = $1 AND status = 'pending_review' AND COALESCE(context_sha256, '') = $5`
	return s.cas(ctx, q, id, imageRef, reviewedBy, at, digest)
}

// SetContextDigest records an upload's digest, only while the row is pending.
func (s *PGStore) SetContextDigest(ctx context.Context, id, digest string) (bool, error) {
	const q = `UPDATE image_submissions SET context_sha256 = $2
		WHERE id = $1 AND status = 'pending_review'`
	return s.cas(ctx, q, id, digest)
}

// RejectSubmission is the reject CAS, mirroring ApproveSubmission.
func (s *PGStore) RejectSubmission(ctx context.Context, id, reviewedBy, reason string, at time.Time) (bool, error) {
	const q = `UPDATE image_submissions
		SET status = 'rejected', reviewed_by = $2, reject_reason = $3, reviewed_at = $4
		WHERE id = $1 AND status = 'pending_review'`
	return s.cas(ctx, q, id, reviewedBy, reason, at)
}

func (s *PGStore) LinkBuild(ctx context.Context, id, buildID string) error {
	const q = `UPDATE image_submissions SET build_id = $2 WHERE id = $1`
	res, err := s.db.ExecContext(ctx, q, id, buildID)
	if err != nil {
		return err
	}
	n, err := res.RowsAffected()
	if err != nil {
		return err
	}
	if n == 0 {
		return ErrNotFound
	}
	return nil
}

// DeleteSubmission is the admin delete: any status, one row, no predicate beyond
// the id. False means the id was already gone (a concurrent delete won).
func (s *PGStore) DeleteSubmission(ctx context.Context, id string) (bool, error) {
	const q = `DELETE FROM image_submissions WHERE id = $1`
	return s.cas(ctx, q, id)
}

// DeletePendingSubmission is the withdraw CAS: owner + pending_review must both
// still hold, so a reviewed submission can never be deleted through this path.
func (s *PGStore) DeletePendingSubmission(ctx context.Context, id, submittedBy string) (bool, error) {
	const q = `DELETE FROM image_submissions
		WHERE id = $1 AND submitted_by = $2 AND status = 'pending_review'`
	return s.cas(ctx, q, id, submittedBy)
}

func (s *PGStore) querySubmissions(ctx context.Context, q string, args ...any) ([]Submission, error) {
	rows, err := s.db.QueryContext(ctx, q, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Submission
	for rows.Next() {
		sub, err := scanSubmissionRows(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, *sub)
	}
	return out, rows.Err()
}

// rowScanner is the read surface shared by *sql.Row (single) and *sql.Rows (in a
// list loop), so one scan body serves both without copy-paste.
type rowScanner interface {
	Scan(dest ...any) error
}

// scanSubmission scans a single row, translating no-rows into ErrNotFound.
func scanSubmission(row *sql.Row) (*Submission, error) {
	sub, err := scanSubmissionRows(row)
	if err == sql.ErrNoRows {
		return nil, ErrNotFound
	}
	return sub, err
}

// scanSubmissionRows decodes one row's columns, mapping nullable text/time
// columns through sql.Null* (NULLIF/absent values become zero, omitted in JSON).
func scanSubmissionRows(row rowScanner) (*Submission, error) {
	var (
		sub                                         Submission
		status                                      string
		imageRef, buildID, reviewedBy, rejectReason sql.NullString
		reviewedAt                                  sql.NullTime
		digest                                      sql.NullString
	)
	if err := row.Scan(
		&sub.ID, &sub.SubmittedBy, &sub.DisplayName, &sub.ContextRef, &status,
		&imageRef, &buildID, &reviewedBy, &rejectReason, &sub.CreatedAt, &reviewedAt, &digest,
	); err != nil {
		return nil, err
	}
	sub.Status = Status(status)
	sub.ImageRef = imageRef.String
	sub.BuildID = buildID.String
	sub.ReviewedBy = reviewedBy.String
	sub.RejectReason = rejectReason.String
	sub.ContextSHA256 = digest.String
	if reviewedAt.Valid {
		t := reviewedAt.Time
		sub.ReviewedAt = &t
	}
	return &sub, nil
}