Janady13/missionmatch / internal/platform/connector/connector.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / internal/platform/connector/connector.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	internal/platform/connector/connector.go
Original SHA-256	7798b1d9c12694ee057dd0e998eb280a2ec1785fe304572cd80e5fcf9b92a235
Git blob SHA	5b5729f5ef8c5d9dcb028211b4ec2055b2e87fd0
Reading-copy SHA-256	7798b1d9c12694ee057dd0e998eb280a2ec1785fe304572cd80e5fcf9b92a235

Original source ↗

Every recorded occurrence

These 9 records contain identical original source bytes. Their paths and revisions remain part of the archive.

Original record	Branch or archive	Revision / identifier	Original location
Janady13/missionmatch / internal/platform/connector/connector.go	feature/opportunity-to-outcome	720d47fae476b0d38477ae4401697d372b1ea2bc	Open original
Janady13/missionmatch / internal/platform/connector/connector.go	collision-deconfliction	4ee5a09972608ed89f8377c6665542d5f450f0a6	Open original
Janady13/missionmatch / internal/platform/connector/connector.go	master	1e9249840a4302ec82e123b9c9cb6f75badea3e4	Open original
Janady13/missionmatch / internal/platform/connector/connector.go	time-to-fill-survival	b66e1e505863d4f17741a4b9ede73509e45bd6c6	Open original
Janady13/missionmatch / internal/platform/connector/connector.go	presentation-readiness	eb574acaa29d71ce4d0f852572e87c275c79fdb7	Open original
Janady13/missionmatch / internal/platform/connector/connector.go	vendor-intelligence	fe9796a19c261951d3f2565206c0c1b5ac294465	Open original
Janady13/missionmatch / internal/platform/connector/connector.go	phase2-connectors	e038d4c687aec16f8bd9bc4d907eea14d91f8704	Open original
Janady13/missionmatch / internal/platform/connector/connector.go	feature/aioe-upgrade	2178e7298d0e5d9af83afa12402fd587f0ba6ed8	Open original
Janady13/missionmatch / internal/platform/connector/connector.go	capture-readiness	bae6af7cb2997e819174a9e9cfff5e16067a1975	Open original

Recovered source transcript

Download complete source text ↓162 lines · 7.0 KB

// 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
}