descrybe/apps/api/internal/processing/orphan_cleanup.go

package processing

import (
	"context"
	"fmt"

	"github.com/descrybe/descrybe-v2/apps/api/internal/billing"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
)

// OrphanProcessedSample is a short diagnostic row for admin report responses.
type OrphanProcessedSample struct {
	ProcessedID  uuid.UUID  `json:"processed_id"`
	CompanyID    uuid.UUID  `json:"company_id"`
	RawProductID *uuid.UUID `json:"raw_product_id,omitempty"`
	Reason       string     `json:"reason"`
	RawStatus    *string    `json:"raw_processing_status,omitempty"`
	RawProcessed *bool      `json:"raw_is_processed,omitempty"`
}

// OrphanProcessedResult is the report (and optional delete) outcome for
// processed_products whose linked raw is missing or unprocessed.
//
// ASSUMPTION: orphans are catalog rows that should not exist while the raw queue
// says unprocessed (or the raw row is gone / SET NULL). Mid-job "processing"
// status is not treated as orphan. Prefer report then delete behind admin
// confirm — no goose data-destroy migration.
//
// Report SQL (ops):
//
//	SELECT p.id, p.company_id, p.raw_product_id, r.processing_status, r.is_processed
//	FROM processed_products p
//	LEFT JOIN raw_products r ON r.id = p.raw_product_id
//	WHERE p.raw_product_id IS NULL
//	   OR r.id IS NULL
//	   OR r.processing_status = 'unprocessed'
//	   OR r.is_processed = false;
//
// Delete SQL (ops, after report):
//
//	DELETE FROM processed_products p
//	WHERE p.raw_product_id IS NULL
//	   OR NOT EXISTS (SELECT 1 FROM raw_products r WHERE r.id = p.raw_product_id)
//	   OR EXISTS (
//	        SELECT 1 FROM raw_products r
//	        WHERE r.id = p.raw_product_id
//	          AND (r.processing_status = 'unprocessed' OR r.is_processed = false)
//	   );
type OrphanProcessedResult struct {
	MissingRaw     int64                   `json:"missing_raw"`
	UnprocessedRaw int64                   `json:"unprocessed_raw"`
	Total          int64                   `json:"total"`
	Deleted        int64                   `json:"deleted"`
	Confirmed      bool                    `json:"confirmed"`
	Samples        []OrphanProcessedSample `json:"samples"`
}

const orphanProcessedSampleLimit = 25

// orphanProcessedWhere matches catalog rows whose raw is missing or unprocessed.
const orphanProcessedWhere = `
	p.raw_product_id IS NULL
	OR r.id IS NULL
	OR r.processing_status = 'unprocessed'
	OR r.is_processed = false`

// ReportOrphanProcessed counts and samples stale catalog rows without deleting.
func ReportOrphanProcessed(ctx context.Context, pool *pgxpool.Pool) (OrphanProcessedResult, error) {
	var out OrphanProcessedResult
	if pool == nil {
		return out, fmt.Errorf("orphan processed: nil pool")
	}
	if err := countOrphanProcessed(ctx, pool, &out); err != nil {
		return out, err
	}
	samples, err := sampleOrphanProcessed(ctx, pool, orphanProcessedSampleLimit)
	if err != nil {
		return out, err
	}
	out.Samples = samples
	return out, nil
}

// CleanupOrphanProcessed reports orphans and, when confirm is true, deletes them.
// processing_job_products.processed_product_id is ON DELETE SET NULL.
//
// Fail-closed: confirm with zero orphans returns ErrOrphanCleanupEmpty (no delete).
// A1 protection: confirm refuses with ErrOrphanCleanupA1Protected when any orphan
// row belongs to the A1 cohort (immutable legacy_company_id).
func CleanupOrphanProcessed(ctx context.Context, pool *pgxpool.Pool, confirm bool) (OrphanProcessedResult, error) {
	out, err := ReportOrphanProcessed(ctx, pool)
	if err != nil {
		return out, err
	}
	out.Confirmed = confirm
	if !confirm {
		return out, nil
	}
	if out.Total == 0 {
		return out, ErrOrphanCleanupEmpty
	}
	touchesA1, err := orphanProcessedTouchesA1(ctx, pool)
	if err != nil {
		return out, err
	}
	if touchesA1 {
		return out, ErrOrphanCleanupA1Protected
	}

	ct, err := pool.Exec(ctx, `
		DELETE FROM processed_products
		WHERE id IN (
			SELECT p.id
			FROM processed_products p
			LEFT JOIN raw_products r ON r.id = p.raw_product_id
			WHERE `+orphanProcessedWhere+`
		)`)
	if err != nil {
		return out, fmt.Errorf("orphan processed delete: %w", err)
	}
	out.Deleted = ct.RowsAffected()
	// Refresh counts after delete so response reflects remaining drift.
	if err := countOrphanProcessed(ctx, pool, &out); err != nil {
		return out, err
	}
	out.Samples = nil
	if remaining, err := sampleOrphanProcessed(ctx, pool, orphanProcessedSampleLimit); err == nil {
		out.Samples = remaining
	}
	return out, nil
}

// orphanProcessedTouchesA1 reports whether any orphan row belongs to A1 cohort.
func orphanProcessedTouchesA1(ctx context.Context, pool *pgxpool.Pool) (bool, error) {
	var n int64
	err := pool.QueryRow(ctx, `
		SELECT COUNT(*)::bigint
		FROM processed_products p
		LEFT JOIN raw_products r ON r.id = p.raw_product_id
		INNER JOIN companies c ON c.id = p.company_id
		WHERE (`+orphanProcessedWhere+`)
		  AND lower(trim(coalesce(c.legacy_company_id, ''))) = lower($1)`,
		billing.A1LegacyCompanyID).Scan(&n)
	if err != nil {
		return false, fmt.Errorf("orphan processed A1 guard: %w", err)
	}
	return n > 0, nil
}

func countOrphanProcessed(ctx context.Context, pool *pgxpool.Pool, out *OrphanProcessedResult) error {
	err := pool.QueryRow(ctx, `
		SELECT
			COUNT(*) FILTER (
				WHERE p.raw_product_id IS NULL OR r.id IS NULL
			)::bigint,
			COUNT(*) FILTER (
				WHERE r.id IS NOT NULL
				  AND (r.processing_status = 'unprocessed' OR r.is_processed = false)
			)::bigint,
			COUNT(*)::bigint
		FROM processed_products p
		LEFT JOIN raw_products r ON r.id = p.raw_product_id
		WHERE `+orphanProcessedWhere).Scan(&out.MissingRaw, &out.UnprocessedRaw, &out.Total)
	if err != nil {
		return fmt.Errorf("orphan processed count: %w", err)
	}
	return nil
}

func sampleOrphanProcessed(ctx context.Context, pool *pgxpool.Pool, limit int) ([]OrphanProcessedSample, error) {
	if limit < 1 {
		limit = orphanProcessedSampleLimit
	}
	rows, err := pool.Query(ctx, `
		SELECT
			p.id,
			p.company_id,
			p.raw_product_id,
			r.processing_status,
			r.is_processed,
			CASE
				WHEN p.raw_product_id IS NULL OR r.id IS NULL THEN 'missing_raw'
				ELSE 'unprocessed_raw'
			END AS reason
		FROM processed_products p
		LEFT JOIN raw_products r ON r.id = p.raw_product_id
		WHERE `+orphanProcessedWhere+`
		ORDER BY p.updated_at DESC NULLS LAST, p.id
		LIMIT $1`, limit)
	if err != nil {
		return nil, fmt.Errorf("orphan processed sample: %w", err)
	}
	defer rows.Close()

	out := make([]OrphanProcessedSample, 0, limit)
	for rows.Next() {
		var s OrphanProcessedSample
		if err := rows.Scan(
			&s.ProcessedID,
			&s.CompanyID,
			&s.RawProductID,
			&s.RawStatus,
			&s.RawProcessed,
			&s.Reason,
		); err != nil {
			return nil, err
		}
		out = append(out, s)
	}
	return out, rows.Err()
}