package connector_events

import (
	"context"
	"database/sql"
	"errors"
	"fmt"

	wq "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
)

// RequireSourceProjectionProtocol checks the exact pack before a standalone
// worker becomes active. Admission repeats the immutable receipt read under
// the pack lock; claim validates the selected range itself.
func (s *Service) RequireSourceProjectionProtocol(ctx context.Context, packID string) error {
	if s == nil || s.workqueue == nil {
		return wq.ErrProjectionProtocolRequired
	}
	if err := s.workqueue.VerifyWorkItemSchema(ctx); err != nil {
		return err
	}
	if _, err := s.workqueue.ProjectionLegacyMaxOrdinal(ctx, packID); err != nil {
		return err
	}
	return requireNativeProjectionWork(ctx, s.db, packID)
}

func requireNativeProjectionWork(ctx context.Context, db *sql.DB, packID string) error {
	var invalid string
	err := db.QueryRowContext(ctx, `SELECT WorkId FROM _System_ConnectorWorkItem
 WHERE InstalledPackId=? AND WorkId LIKE 'csp:%' AND State IN (`+wq.NonTerminalWorkStateSQLList+`)
 AND (InputStartOrdinal IS NULL OR InputStartOrdinal<=0 OR LaneOrdinal<InputStartOrdinal) LIMIT 1`, packID).Scan(&invalid)
	if err == nil {
		return fmt.Errorf("%w: old-format or invalid WorkItem %q; drain old work before cutover", wq.ErrProjectionProtocolRequired, invalid)
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return err
	}
	return nil
}

// RequireSourceProjectionProtocol is a process-activation gate. Admission and
// replay additionally validate the exact incarnation in their pack transaction.
// There is no per-poll or per-Event history scan.
func RequireSourceProjectionProtocol(ctx context.Context, db *sql.DB) error {
	if db == nil {
		return fmt.Errorf("%w: database is required", wq.ErrProjectionProtocolRequired)
	}
	if err := wq.NewRepository(db, nil, nil).VerifyWorkItemSchema(ctx); err != nil {
		return err
	}
	var invalid string
	err := db.QueryRowContext(ctx, `SELECT p.PackInstallId FROM _System_InstalledPack p
 LEFT JOIN _System_ConnectorProjectionProtocol protocol ON protocol.InstalledPackId=p.PackInstallId
 WHERE p.Status='active' AND COALESCE(p.IsDeleted,0)=0
 AND (EXISTS(SELECT 1 FROM _System_ConnectorInboundRoute r WHERE r.InstalledPackId=p.PackInstallId AND COALESCE(r.IsDeleted,0)=0)
 OR EXISTS(SELECT 1 FROM _System_ConnectorEvent e WHERE e.InstalledPackId=p.PackInstallId)
 OR EXISTS(SELECT 1 FROM _System_ConnectorWorkItem w WHERE w.InstalledPackId=p.PackInstallId))
 AND (protocol.InstalledPackId IS NULL OR protocol.ProtocolVersion<>? OR protocol.LegacyMaxOrdinal<0) LIMIT 1`, wq.ProjectionRangeProtocolVersion).Scan(&invalid)
	if err == nil {
		return fmt.Errorf("%w: pack %q", wq.ErrProjectionProtocolRequired, invalid)
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return err
	}
	err = db.QueryRowContext(ctx, `SELECT w.WorkId FROM _System_ConnectorWorkItem w
 JOIN _System_InstalledPack p ON p.PackInstallId=w.InstalledPackId AND p.Status='active' AND COALESCE(p.IsDeleted,0)=0
 WHERE w.WorkId LIKE 'csp:%' AND w.State IN (`+wq.NonTerminalWorkStateSQLList+`)
 AND (w.InputStartOrdinal IS NULL OR w.InputStartOrdinal<=0 OR w.LaneOrdinal<w.InputStartOrdinal) LIMIT 1`).Scan(&invalid)
	if err == nil {
		return fmt.Errorf("%w: old-format or invalid WorkItem %q; drain old work before cutover", wq.ErrProjectionProtocolRequired, invalid)
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return err
	}
	return nil
}