// Package connector flows a consenting pilot tenant's live workforce data from an enterprise ATS or HRIS into
// the candidates table, incrementally, behind a small swappable Connector interface so iCIMS, Workday, or
// another AUTHORIZED system is configuration not code, exactly like the storage and inference ports. It reuses
// the proven sourcing ingest path (sourcing.IngestPerson), so a connector record becomes a real structured
// scoring input (skills, certs, agencies as rows) and is deduplicated against the existing bench, and it never
// writes a clearance value (sourced clearance is unknown, never a negative or a held clearance). Each record
// gets a sourcing_log provenance row in the same transaction, the same compliance promise the import path keeps.
//
// It runs as the BYPASSRLS migrator role with an explicit organization_id, exactly like cmd/import-candidates,
// because the ingest path writes the global skills ontology; RLS is not relied on, the org is passed on every
// write and filtered explicitly on every read.
//
// CUI SAFETY, by construction: only a FedRAMP relevant system's adapter is admitted here. A cloud unified
// aggregator (Merge.dev) is NOT FedRAMP authorized, so no Merge adapter is ever constructed on this CUI core
// path; a Merge style adapter may live behind the same interface later ONLY for a synthetic demo tenant, never
// for cleared workforce data. Secrets (OAuth client id and secret) load from the server side env only.
package connector

import (
	"context"
	"encoding/json"
	"fmt"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/missionworkforce/missionmatch/internal/platform/database/dbgen"
	"github.com/missionworkforce/missionmatch/internal/platform/extraction"
	"github.com/missionworkforce/missionmatch/internal/platform/sourcing"
)

// Connector pulls candidate records from an external system incrementally. An adapter maps the system's records
// to the shared provider independent SourcedPerson shape; the sync orchestration ingests them. FetchPage pulls
// one page of records updated at or after since, using an opaque page cursor (empty for the first page), and
// returns the mapped records, the next page cursor (empty when the last page), and the max updated timestamp
// seen on the page (to advance the incremental watermark). A 429 maps to sourcing.ErrThrottled so the caller
// stops gracefully within the provider's daily cap.
type Connector interface {
	Name() string
	FetchPage(ctx context.Context, since time.Time, page string) (people []sourcing.SourcedPerson, nextPage string, maxUpdated time.Time, err error)
}

// SyncResult summarizes one sync run: the records ingested, the pages walked, and the advanced watermark.
type SyncResult struct {
	Connector string
	Synced    int
	Pages     int
	Watermark time.Time
}

// Sync runs one INCREMENTAL sync for a connector and one org. It reads the org's watermark, pages through the
// records updated since it (bounded by maxRecords so a large backfill degrades gracefully within the provider
// cap), and ingests each page as structured candidates plus a sourcing_log provenance row through the proven
// sourcing path, advancing and persisting the watermark per page in one atomic transaction. Network I/O
// (FetchPage) runs OUTSIDE the transaction so a page fetch never holds one open. A throttle or a fetch error
// stops the run gracefully, keeping whatever pages already committed. No PII is logged.
func Sync(ctx context.Context, pool *pgxpool.Pool, orgID uuid.UUID, c Connector, ext extraction.Extractor, legalBasis string, maxRecords int) (SyncResult, error) {
	if pool == nil {
		return SyncResult{}, fmt.Errorf("connector: no database")
	}
	if maxRecords <= 0 {
		maxRecords = 1000
	}
	res := SyncResult{Connector: c.Name()}

	since, err := readWatermark(ctx, pool, orgID, c.Name())
	if err != nil {
		return SyncResult{}, err
	}
	res.Watermark = since

	page := ""
	for res.Synced < maxRecords {
		people, nextPage, maxUpdated, ferr := c.FetchPage(ctx, since, page)
		if ferr != nil {
			return res, ferr
		}
		res.Pages++

		if len(people) > 0 {
			if maxUpdated.After(res.Watermark) {
				res.Watermark = maxUpdated
			}
			added, e := ingestPage(ctx, pool, orgID, c.Name(), ext, legalBasis, people, res.Watermark)
			if e != nil {
				return res, e
			}
			res.Synced += added
		}

		if nextPage == "" {
			break
		}
		page = nextPage
	}
	return res, nil
}

// ingestPage writes one page of records as structured candidates plus their sourcing_log provenance rows and
// advances the watermark, all in ONE transaction, so a page is atomic: either every record and the cursor
// advance commit together or none do. It returns the number of records ingested.
func ingestPage(ctx context.Context, pool *pgxpool.Pool, orgID uuid.UUID, name string, ext extraction.Extractor, legalBasis string, people []sourcing.SourcedPerson, watermark time.Time) (int, error) {
	added := 0
	err := pgx.BeginFunc(ctx, pool, func(tx pgx.Tx) error {
		q := dbgen.New(tx)
		queryJSON, _ := json.Marshal(map[string]any{"connector": name})
		for _, sp := range people {
			candID, e := sourcing.IngestPerson(ctx, q, orgID, sp, ext)
			if e != nil {
				return e
			}
			if e := q.InsertSourcingLog(ctx, dbgen.InsertSourcingLogParams{
				OrganizationID: orgID,
				CandidateID:    &candID,
				ExternalID:     sp.ExternalID,
				Source:         "ATS",
				Provider:       name,
				Query:          string(queryJSON),
				LegalBasis:     legalBasis,
			}); e != nil {
				return fmt.Errorf("insert sourcing log: %w", e)
			}
			added++
		}
		return advanceWatermark(ctx, tx, orgID, name, watermark, added)
	})
	if err != nil {
		return 0, err
	}
	return added, nil
}

// readWatermark returns the org's incremental floor for a connector, or the zero epoch when no cursor exists yet
// (the first run pulls the full history the provider window allows). It filters org explicitly (the migrator
// bypasses RLS).
func readWatermark(ctx context.Context, pool *pgxpool.Pool, orgID uuid.UUID, name string) (time.Time, error) {
	var wm time.Time
	err := pool.QueryRow(ctx,
		`SELECT watermark FROM connector_sync_state WHERE organization_id = $1 AND connector_name = $2`,
		orgID, name).Scan(&wm)
	if err == pgx.ErrNoRows {
		return time.Unix(0, 0).UTC(), nil
	}
	return wm, err
}

// advanceWatermark upserts the org's cursor for a connector to the new watermark (never lowering it) and bumps
// the run bookkeeping, inside the caller's transaction so it commits atomically with the page's ingests.
func advanceWatermark(ctx context.Context, tx pgx.Tx, orgID uuid.UUID, name string, watermark time.Time, added int) error {
	_, err := tx.Exec(ctx, `
		INSERT INTO connector_sync_state (organization_id, connector_name, watermark, last_run_at, records_synced)
		VALUES ($1, $2, $3, now(), $4)
		ON CONFLICT (organization_id, connector_name) DO UPDATE
		SET watermark      = GREATEST(connector_sync_state.watermark, EXCLUDED.watermark),
		    last_run_at    = now(),
		    records_synced = connector_sync_state.records_synced + EXCLUDED.records_synced`,
		orgID, name, watermark, added)
	return err
}