rohit/probo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Unify enrichment tracking and outcome-based status

Browse Source 
Make common_tracker_patterns and common_third_parties share one
enrichment-tracking model and fix the misleading proboctl status.

Both tables now carry the enrichment JSONB provenance payload, an
enrichment_attempts counter, and a last_enrichment_attempt_at clock.
On common_tracker_patterns the enriched_at done-flag is renamed to
last_enrichment_attempt_at and stamped at claim time, so it is truthful
to "attempt" rather than "success". A row is considered to have been
through the workflow when it carries an enrichment payload, not when a
timestamp is set, which lets stale recovery key off the payload being
absent with budget remaining, exactly like common_third_parties.

The claim path reads the attempt counter and timestamp back via
RETURNING so the in-memory receiver matches the database clock instead
of a separate app-side time.Now.

The enricher builds a per-field provenance payload (description and
third-party outcomes plus the mapping attribution) and persists it via
UpdateEnrichment, named to mirror the common-third-party sibling. The
common pattern enrichment worker gains a max-attempts ceiling so a
permanently failing row stops looping.

proboctl now shows "enriched" only when every field the last run
recorded an outcome for resolved a value, otherwise "partial (X/Y)",
replacing the misleading "enriched (no description)" label.

Signed-off-by: Émile Ré <emile@probo.com>
This commit is contained in:
Émile Ré
2026-06-15 16:25:53 +02:00
parent a6ed64caaf
commit ef6cead482
17 changed files with 705 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
pkg/coredata/common_third_party.go
View File

@@ -53,6 +53,7 @@ type (
EnrichmentRequestedAt *time.Time `db:"enrichment_requested_at"`
Enrichment json.RawMessage `db:"enrichment"`
EnrichmentAttempts int `db:"enrichment_attempts"`
LastEnrichmentAttemptAt *time.Time `db:"last_enrichment_attempt_at"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
@@ -131,6 +132,7 @@ SELECT
enrichment_requested_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -191,6 +193,7 @@ SELECT
enrichment_requested_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -251,6 +254,7 @@ SELECT
enrichment_requested_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -310,6 +314,7 @@ INSERT INTO common_third_parties (
enrichment_requested_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
) VALUES (
@@ -335,6 +340,7 @@ INSERT INTO common_third_parties (
@enrichment_requested_at,
@enrichment,
@enrichment_attempts,
@last_enrichment_attempt_at,
@created_at,
@updated_at
)
@@ -363,6 +369,7 @@ INSERT INTO common_third_parties (
"enrichment_requested_at": t.EnrichmentRequestedAt,
"enrichment": t.Enrichment,
"enrichment_attempts": t.EnrichmentAttempts,
"last_enrichment_attempt_at": t.LastEnrichmentAttemptAt,
"created_at": t.CreatedAt,
"updated_at": t.UpdatedAt,
}
@@ -406,6 +413,7 @@ INSERT INTO common_third_parties (
enrichment_requested_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
) VALUES (
@@ -431,6 +439,7 @@ INSERT INTO common_third_parties (
@enrichment_requested_at,
@enrichment,
@enrichment_attempts,
@last_enrichment_attempt_at,
@created_at,
@updated_at
)
@@ -476,6 +485,7 @@ RETURNING
enrichment_requested_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
`
@@ -505,6 +515,7 @@ RETURNING
"enrichment_requested_at": t.EnrichmentRequestedAt,
"enrichment": t.Enrichment,
"enrichment_attempts": t.EnrichmentAttempts,
"last_enrichment_attempt_at": t.LastEnrichmentAttemptAt,
"created_at": t.CreatedAt,
"updated_at": t.UpdatedAt,
}
@@ -571,6 +582,7 @@ SELECT
enrichment_requested_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -625,6 +637,7 @@ SELECT
enrichment_requested_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -769,6 +782,7 @@ SELECT
enrichment_requested_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -862,6 +876,7 @@ SELECT
enrichment_requested_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -895,8 +910,8 @@ LIMIT 1;
}
// ClearEnrichmentRequestedAt removes the row from the enrichment queue
// and bumps the attempt counter. It bumps updated_at so the
// stale-recovery clock starts at claim time, keeping
// and bumps the attempt counter. It stamps last_enrichment_attempt_at so
// the stale-recovery clock starts at claim time, keeping
// ResetStaleCommonThirdPartyEnrichments from re-arming a row that is
// still being processed. The attempt counter is incremented up front so
// a crash between claim and persist still counts against the retry
@@ -910,19 +925,27 @@ UPDATE common_third_parties
SET
enrichment_requested_at = NULL,
enrichment_attempts = enrichment_attempts + 1,
last_enrichment_attempt_at = NOW(),
updated_at = NOW()
WHERE id = @id
RETURNING enrichment_attempts, last_enrichment_attempt_at
`
args := pgx.StrictNamedArgs{"id": t.ID}
_, err := tx.Exec(ctx, q, args)
var (
attempts int
lastAttempt *time.Time
)
err := tx.QueryRow(ctx, q, args).Scan(&attempts, &lastAttempt)
if err != nil {
return fmt.Errorf("cannot clear enrichment requested at: %w", err)
}
t.EnrichmentRequestedAt = nil
t.EnrichmentAttempts++
t.EnrichmentAttempts = attempts
t.LastEnrichmentAttemptAt = lastAttempt
return nil
}
@@ -1024,7 +1047,7 @@ WHERE
AND enrichment IS NULL
AND enrichment_attempts > 0
AND enrichment_attempts < @max_attempts
AND updated_at < @stale_before
AND last_enrichment_attempt_at < @stale_before
`
args := pgx.StrictNamedArgs{

7
pkg/coredata/common_third_party_filter.go
View File

@@ -23,10 +23,9 @@ import (
// CommonThirdPartyEnrichmentState is a synthetic filter over the
// enrichment_requested_at / enrichment columns. It is not a stored
// column; it classifies a row's position in the enrichment lifecycle.
// Unlike common tracker patterns, a common third party has no
// enriched_at column: a row is "enriched" once it carries an enrichment
// payload (Process always writes one, even on a no-result run).
// column; it classifies a row's position in the enrichment lifecycle. A
// row is "enriched" once it carries an enrichment payload (Process always
// writes one, even on a no-result run).
type CommonThirdPartyEnrichmentState string
const (

201
pkg/coredata/common_tracker_pattern.go
View File

@@ -16,6 +16,7 @@ package coredata
import (
"context"
"encoding/json"
"errors"
"fmt"
"maps"
@@ -29,18 +30,20 @@ import (
type (
CommonTrackerPattern struct {
ID gid.GID `db:"id"`
CommonThirdPartyID *gid.GID `db:"common_third_party_id"`
TrackerType TrackerType `db:"tracker_type"`
Pattern string `db:"pattern"`
MatchType TrackerPatternMatchType `db:"match_type"`
Description string `db:"description"`
MaxAgeSeconds *int `db:"max_age_seconds"`
Confidence float32 `db:"confidence"`
EnrichmentRequestedAt *time.Time `db:"enrichment_requested_at"`
EnrichedAt *time.Time `db:"enriched_at"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
ID gid.GID `db:"id"`
CommonThirdPartyID *gid.GID `db:"common_third_party_id"`
TrackerType TrackerType `db:"tracker_type"`
Pattern string `db:"pattern"`
MatchType TrackerPatternMatchType `db:"match_type"`
Description string `db:"description"`
MaxAgeSeconds *int `db:"max_age_seconds"`
Confidence float32 `db:"confidence"`
EnrichmentRequestedAt *time.Time `db:"enrichment_requested_at"`
Enrichment json.RawMessage `db:"enrichment"`
EnrichmentAttempts int `db:"enrichment_attempts"`
LastEnrichmentAttemptAt *time.Time `db:"last_enrichment_attempt_at"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
CommonTrackerPatterns []*CommonTrackerPattern
@@ -62,7 +65,9 @@ SELECT
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -111,7 +116,9 @@ SELECT
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -163,7 +170,9 @@ INSERT INTO common_tracker_patterns (
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
) VALUES (
@@ -176,25 +185,29 @@ INSERT INTO common_tracker_patterns (
@max_age_seconds,
@confidence,
@enrichment_requested_at,
@enriched_at,
@enrichment,
@enrichment_attempts,
@last_enrichment_attempt_at,
@created_at,
@updated_at
)
`
args := pgx.StrictNamedArgs{
"id": p.ID,
"common_third_party_id": p.CommonThirdPartyID,
"tracker_type": p.TrackerType,
"pattern": p.Pattern,
"match_type": p.MatchType,
"description": p.Description,
"max_age_seconds": p.MaxAgeSeconds,
"confidence": p.Confidence,
"enrichment_requested_at": p.EnrichmentRequestedAt,
"enriched_at": p.EnrichedAt,
"created_at": p.CreatedAt,
"updated_at": p.UpdatedAt,
"id": p.ID,
"common_third_party_id": p.CommonThirdPartyID,
"tracker_type": p.TrackerType,
"pattern": p.Pattern,
"match_type": p.MatchType,
"description": p.Description,
"max_age_seconds": p.MaxAgeSeconds,
"confidence": p.Confidence,
"enrichment_requested_at": p.EnrichmentRequestedAt,
"enrichment": p.Enrichment,
"enrichment_attempts": p.EnrichmentAttempts,
"last_enrichment_attempt_at": p.LastEnrichmentAttemptAt,
"created_at": p.CreatedAt,
"updated_at": p.UpdatedAt,
}
_, err := conn.Exec(ctx, q, args)
@@ -226,7 +239,9 @@ INSERT INTO common_tracker_patterns (
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
) VALUES (
@@ -240,6 +255,8 @@ INSERT INTO common_tracker_patterns (
@confidence,
CASE WHEN @description = '' THEN NOW() ELSE NULL END,
NULL,
0,
NULL,
@created_at,
@updated_at
)
@@ -255,8 +272,8 @@ SET
-- A blank, unlinked catalog row that now gains a third party is
-- re-queued for enrichment: the enrichment agent leaves descriptions
-- blank when it cannot substantiate a purpose, and knowing the vendor
-- gives it a second, better-informed attempt. enriched_at is cleared
-- so the row is no longer terminal.
-- gives it a second, better-informed attempt. The attempt counter is
-- reset so the re-armed row gets a fresh retry budget.
enrichment_requested_at = CASE
WHEN common_tracker_patterns.description = ''
AND common_tracker_patterns.common_third_party_id IS NULL
@@ -264,12 +281,12 @@ SET
THEN NOW()
ELSE common_tracker_patterns.enrichment_requested_at
END,
enriched_at = CASE
enrichment_attempts = CASE
WHEN common_tracker_patterns.description = ''
AND common_tracker_patterns.common_third_party_id IS NULL
AND EXCLUDED.common_third_party_id IS NOT NULL
THEN NULL
ELSE common_tracker_patterns.enriched_at
THEN 0
ELSE common_tracker_patterns.enrichment_attempts
END,
updated_at = EXCLUDED.updated_at
RETURNING
@@ -282,7 +299,9 @@ RETURNING
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
`
@@ -352,7 +371,9 @@ SELECT
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -469,7 +490,9 @@ SELECT
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -516,7 +539,9 @@ SELECT
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -548,8 +573,11 @@ LIMIT 1;
return nil
}
// ClearEnrichmentRequestedAt removes the row from the enrichment queue. It
// bumps updated_at so the stale-recovery clock starts at claim time.
// ClearEnrichmentRequestedAt removes the row from the enrichment queue and
// records the attempt: it bumps the attempt counter and stamps
// last_enrichment_attempt_at, which is the stale-recovery clock. The
// attempt counter is incremented up front so a crash between claim and
// persist still counts against the retry budget. It bumps updated_at too.
func (p *CommonTrackerPattern) ClearEnrichmentRequestedAt(
ctx context.Context,
tx pg.Tx,
@@ -558,42 +586,57 @@ func (p *CommonTrackerPattern) ClearEnrichmentRequestedAt(
UPDATE common_tracker_patterns
SET
enrichment_requested_at = NULL,
enrichment_attempts = enrichment_attempts + 1,
last_enrichment_attempt_at = NOW(),
updated_at = NOW()
WHERE id = @id
RETURNING enrichment_attempts, last_enrichment_attempt_at
`
args := pgx.StrictNamedArgs{"id": p.ID}
_, err := tx.Exec(ctx, q, args)
var (
attempts int
lastAttempt *time.Time
)
err := tx.QueryRow(ctx, q, args).Scan(&attempts, &lastAttempt)
if err != nil {
return fmt.Errorf("cannot clear enrichment requested at: %w", err)
}
p.EnrichmentRequestedAt = nil
p.EnrichmentAttempts = attempts
p.LastEnrichmentAttemptAt = lastAttempt
return nil
}
// SetEnriched records the researched description and marks the row
// enriched so the stale-recovery loop never re-queues it. An empty
// description is allowed: the enrichment agent leaves it blank when it
// cannot substantiate a purpose, and a later third-party link re-arms
// enrichment for a second attempt. When thirdPartyID is non-nil it links
// the row to that third party, but only when none is set yet
// (COALESCE) — the enrichment worker links, it never overrides an
// attribution the mapping pipeline already resolved.
func (p *CommonTrackerPattern) SetEnriched(
// UpdateEnrichment records the researched description and the per-run
// enrichment provenance payload (named to mirror
// CommonThirdParty.UpdateEnrichment, the sibling persist step). The
// payload presence is what marks a row as having been through the
// workflow, so the stale-recovery loop never re-queues it
// (last_enrichment_attempt_at, the attempt clock, is stamped separately at
// claim time). An empty description is allowed: the enrichment agent
// leaves it blank when it cannot substantiate a purpose, and a later
// third-party link re-arms enrichment for a second attempt. When
// thirdPartyID is non-nil it links the row to that third party, but only
// when none is set yet (COALESCE) — the enrichment worker links, it never
// overrides an attribution the mapping pipeline already resolved.
func (p *CommonTrackerPattern) UpdateEnrichment(
ctx context.Context,
tx pg.Tx,
description string,
thirdPartyID *gid.GID,
enrichment json.RawMessage,
) error {
q := `
UPDATE common_tracker_patterns
SET
description = @description,
common_third_party_id = COALESCE(common_third_party_id, @third_party_id),
enriched_at = NOW(),
enrichment = @enrichment,
enrichment_requested_at = NULL,
updated_at = NOW()
WHERE id = @id
@@ -603,6 +646,7 @@ WHERE id = @id
"id": p.ID,
"description": description,
"third_party_id": thirdPartyID,
"enrichment": enrichment,
}
result, err := tx.Exec(ctx, q, args)
@@ -615,6 +659,8 @@ WHERE id = @id
}
p.Description = description
p.Enrichment = enrichment
p.EnrichmentRequestedAt = nil
if p.CommonThirdPartyID == nil {
p.CommonThirdPartyID = thirdPartyID
@@ -624,13 +670,21 @@ WHERE id = @id
}
// ResetStaleEnrichments re-queues rows whose enrichment was claimed but
// never completed (no enriched_at, still description-less) and have been
// idle longer than staleAfter, so a crashed or timed-out enrichment is
// retried.
// never completed and have been idle longer than staleAfter, so a crashed
// or timed-out enrichment is retried.
//
// A claimed row has enrichment_attempts > 0 (the claim increments it) and
// a completed row carries a non-null enrichment payload (UpdateEnrichment
// always writes it, even on a blank-description run), so the sweep targets
// rows that were claimed but carry no payload yet. Curated rows that were
// never enqueued keep enrichment_attempts = 0 and are left untouched. The
// max-attempts ceiling stops permanently failing rows from looping
// forever. last_enrichment_attempt_at, stamped at claim, is the idle clock.
func ResetStaleEnrichments(
ctx context.Context,
conn pg.Querier,
staleAfter time.Duration,
maxAttempts int,
) error {
q := `
UPDATE common_tracker_patterns
@@ -639,12 +693,16 @@ SET
updated_at = NOW()
WHERE
enrichment_requested_at IS NULL
AND enriched_at IS NULL
AND description = ''
AND updated_at < @stale_before
AND enrichment IS NULL
AND enrichment_attempts > 0
AND enrichment_attempts < @max_attempts
AND last_enrichment_attempt_at < @stale_before
`
args := pgx.StrictNamedArgs{"stale_before": time.Now().Add(-staleAfter)}
args := pgx.StrictNamedArgs{
"max_attempts": maxAttempts,
"stale_before": time.Now().Add(-staleAfter),
}
_, err := conn.Exec(ctx, q, args)
if err != nil {
@@ -670,7 +728,9 @@ SELECT
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -706,12 +766,12 @@ func (p *CommonTrackerPattern) CursorKey(field CommonTrackerPatternOrderField) p
return page.NewCursorKey(p.ID, p.CreatedAt)
case CommonTrackerPatternOrderFieldUpdatedAt:
return page.NewCursorKey(p.ID, p.UpdatedAt)
case CommonTrackerPatternOrderFieldEnrichedAt:
if p.EnrichedAt == nil {
case CommonTrackerPatternOrderFieldLastEnrichmentAttemptAt:
if p.LastEnrichmentAttemptAt == nil {
return page.NewCursorKey(p.ID, time.Time{})
}
return page.NewCursorKey(p.ID, *p.EnrichedAt)
return page.NewCursorKey(p.ID, *p.LastEnrichmentAttemptAt)
}
panic(fmt.Sprintf("unsupported order by: %s", field))
@@ -737,7 +797,9 @@ SELECT
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
enrichment,
enrichment_attempts,
last_enrichment_attempt_at,
created_at,
updated_at
FROM
@@ -874,10 +936,14 @@ WHERE
// RequestEnrichmentByIDs arms enrichment on the given common tracker
// patterns by stamping enrichment_requested_at, which is the only column
// the enrichment worker claims on. Already-enriched rows are re-processed
// too: the worker overwrites enriched_at and the description when it runs.
// Returns the number of rows re-queued. This is the async fallback path;
// the synchronous enricher service is preferred.
// the enrichment worker claims on. It resets enrichment_attempts to 0 so
// the re-queued rows get a fresh retry budget: the claim path bumps the
// counter on every run, and without a reset a row near the max-attempts
// ceiling would not be re-armed by stale recovery if a re-run crashed.
// Already-enriched rows are re-processed too: the worker overwrites the
// description and enrichment payload when it runs. Returns the number of
// rows re-queued. This is the async fallback path; the synchronous
// enricher service is preferred.
func (ps *CommonTrackerPatterns) RequestEnrichmentByIDs(
ctx context.Context,
tx pg.Tx,
@@ -887,6 +953,7 @@ func (ps *CommonTrackerPatterns) RequestEnrichmentByIDs(
UPDATE common_tracker_patterns
SET
enrichment_requested_at = NOW(),
enrichment_attempts = 0,
updated_at = NOW()
WHERE
id = ANY(@ids)

10
pkg/coredata/common_tracker_pattern_filter.go
View File

@@ -22,7 +22,7 @@ import (
)
// CommonTrackerPatternEnrichmentState is a synthetic filter over the
// enrichment_requested_at / enriched_at columns. It is not a stored
// enrichment_requested_at / enrichment columns. It is not a stored
// column; it classifies a row's position in the enrichment lifecycle.
type CommonTrackerPatternEnrichmentState string
@@ -30,8 +30,8 @@ const (
// CommonTrackerPatternEnrichmentStateQueued: a row armed for the
// enrichment worker (enrichment_requested_at IS NOT NULL).
CommonTrackerPatternEnrichmentStateQueued CommonTrackerPatternEnrichmentState = "QUEUED"
// CommonTrackerPatternEnrichmentStateEnriched: a row whose
// enrichment has completed (enriched_at IS NOT NULL) and is not
// CommonTrackerPatternEnrichmentStateEnriched: a row that has been
// through the enrichment workflow (enrichment IS NOT NULL) and is not
// re-queued.
CommonTrackerPatternEnrichmentStateEnriched CommonTrackerPatternEnrichmentState = "ENRICHED"
// CommonTrackerPatternEnrichmentStateUnenriched: a row never enriched
@@ -183,9 +183,9 @@ func (f *CommonTrackerPatternFilter) SQLFragment() string {
CASE
WHEN @filter_state_queued::boolean THEN enrichment_requested_at IS NOT NULL
WHEN @filter_state_enriched::boolean THEN
enrichment_requested_at IS NULL AND enriched_at IS NOT NULL
enrichment_requested_at IS NULL AND enrichment IS NOT NULL
WHEN @filter_state_unenriched::boolean THEN
enrichment_requested_at IS NULL AND enriched_at IS NULL
enrichment_requested_at IS NULL AND enrichment IS NULL
ELSE TRUE
END
)`

18
pkg/coredata/common_tracker_pattern_order_field.go
View File

@@ -24,11 +24,11 @@ import (
type CommonTrackerPatternOrderField string
const (
CommonTrackerPatternOrderFieldPattern CommonTrackerPatternOrderField = "PATTERN"
CommonTrackerPatternOrderFieldConfidence CommonTrackerPatternOrderField = "CONFIDENCE"
CommonTrackerPatternOrderFieldCreatedAt CommonTrackerPatternOrderField = "CREATED_AT"
CommonTrackerPatternOrderFieldUpdatedAt CommonTrackerPatternOrderField = "UPDATED_AT"
CommonTrackerPatternOrderFieldEnrichedAt CommonTrackerPatternOrderField = "ENRICHED_AT"
CommonTrackerPatternOrderFieldPattern CommonTrackerPatternOrderField = "PATTERN"
CommonTrackerPatternOrderFieldConfidence CommonTrackerPatternOrderField = "CONFIDENCE"
CommonTrackerPatternOrderFieldCreatedAt CommonTrackerPatternOrderField = "CREATED_AT"
CommonTrackerPatternOrderFieldUpdatedAt CommonTrackerPatternOrderField = "UPDATED_AT"
CommonTrackerPatternOrderFieldLastEnrichmentAttemptAt CommonTrackerPatternOrderField = "LAST_ENRICHMENT_ATTEMPT_AT"
)
var (
@@ -44,7 +44,7 @@ func CommonTrackerPatternOrderFields() []CommonTrackerPatternOrderField {
CommonTrackerPatternOrderFieldConfidence,
CommonTrackerPatternOrderFieldCreatedAt,
CommonTrackerPatternOrderFieldUpdatedAt,
CommonTrackerPatternOrderFieldEnrichedAt,
CommonTrackerPatternOrderFieldLastEnrichmentAttemptAt,
}
}
@@ -55,7 +55,7 @@ func (v CommonTrackerPatternOrderField) IsValid() bool {
CommonTrackerPatternOrderFieldConfidence,
CommonTrackerPatternOrderFieldCreatedAt,
CommonTrackerPatternOrderFieldUpdatedAt,
CommonTrackerPatternOrderFieldEnrichedAt:
CommonTrackerPatternOrderFieldLastEnrichmentAttemptAt:
return true
}
@@ -91,8 +91,8 @@ func (v CommonTrackerPatternOrderField) Column() string {
return "created_at"
case CommonTrackerPatternOrderFieldUpdatedAt:
return "updated_at"
case CommonTrackerPatternOrderFieldEnrichedAt:
return "COALESCE(enriched_at, '0001-01-01T00:00:00Z'::timestamptz)"
case CommonTrackerPatternOrderFieldLastEnrichmentAttemptAt:
return "COALESCE(last_enrichment_attempt_at, '0001-01-01T00:00:00Z'::timestamptz)"
}
panic(fmt.Sprintf("unsupported order by: %s", v))

170
pkg/coredata/common_tracker_pattern_test.go
View File

@@ -16,6 +16,7 @@ package coredata_test
import (
"context"
"encoding/json"
"testing"
"time"
@@ -100,12 +101,12 @@ func loadCommonTrackerPattern(
return reloaded
}
// TestCommonTrackerPattern_SetEnriched_AllowsEmptyDescription pins the
// no-fabrication contract: the enrichment worker records an empty
// description when it cannot substantiate a purpose, and the row is
// still marked terminally enriched so the stale-recovery loop never
// re-queues it.
func TestCommonTrackerPattern_SetEnriched_AllowsEmptyDescription(t *testing.T) {
// TestCommonTrackerPattern_UpdateEnrichment_AllowsEmptyDescription pins
// the no-fabrication contract: the enrichment worker records an empty
// description when it cannot substantiate a purpose, but still writes an
// enrichment payload so the row reads as having been through the workflow
// and the stale-recovery loop never re-queues it.
func TestCommonTrackerPattern_UpdateEnrichment_AllowsEmptyDescription(t *testing.T) {
t.Parallel()
client := test.PGClient(t)
@@ -126,30 +127,32 @@ func TestCommonTrackerPattern_SetEnriched_AllowsEmptyDescription(t *testing.T) {
}
insertCommonTrackerPattern(t, ctx, client, cp)
payload := json.RawMessage(`{"status":"no_result","fields":{"description":{"status":"not_found"}}}`)
require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
return cp.SetEnriched(ctx, tx, "", nil)
return cp.UpdateEnrichment(ctx, tx, "", nil, payload)
}))
reloaded := loadCommonTrackerPattern(t, ctx, client, cp.ID)
assert.Equal(t, "", reloaded.Description, "blank description must stay blank")
assert.NotNil(t, reloaded.EnrichedAt, "blank row must be marked enriched (terminal-for-now)")
assert.NotEmpty(t, reloaded.Enrichment, "blank row must still record an enrichment payload")
assert.Nil(t, reloaded.EnrichmentRequestedAt, "enriched row must leave the queue")
// A blank but enriched row must NOT be re-queued by stale recovery:
// enriched_at is set, so the stale sweep skips it.
// the enrichment payload is present, so the stale sweep skips it.
require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
return coredata.ResetStaleEnrichments(ctx, conn, 0)
return coredata.ResetStaleEnrichments(ctx, conn, 0, 3)
}))
afterSweep := loadCommonTrackerPattern(t, ctx, client, cp.ID)
assert.Nil(t, afterSweep.EnrichmentRequestedAt, "stale recovery must not re-queue an enriched blank row")
}
// TestCommonTrackerPattern_SetEnriched_LinksThirdPartyWithoutOverride
// TestCommonTrackerPattern_UpdateEnrichment_LinksThirdPartyWithoutOverride
// pins the link-no-override contract: the enrichment worker links a
// resolved third party only when the row has none, and never clobbers an
// attribution the mapping pipeline already resolved.
func TestCommonTrackerPattern_SetEnriched_LinksThirdPartyWithoutOverride(t *testing.T) {
func TestCommonTrackerPattern_UpdateEnrichment_LinksThirdPartyWithoutOverride(t *testing.T) {
t.Parallel()
client := test.PGClient(t)
@@ -173,7 +176,7 @@ func TestCommonTrackerPattern_SetEnriched_LinksThirdPartyWithoutOverride(t *test
insertCommonTrackerPattern(t, ctx, client, cp)
require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
return cp.SetEnriched(ctx, tx, "Analytics tracker.", &party.ID)
return cp.UpdateEnrichment(ctx, tx, "Analytics tracker.", &party.ID, json.RawMessage(`{"status":"done"}`))
}))
reloaded := loadCommonTrackerPattern(t, ctx, client, cp.ID)
@@ -195,7 +198,7 @@ func TestCommonTrackerPattern_SetEnriched_LinksThirdPartyWithoutOverride(t *test
insertCommonTrackerPattern(t, ctx, client, cp)
require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
return cp.SetEnriched(ctx, tx, "Analytics tracker.", &other.ID)
return cp.UpdateEnrichment(ctx, tx, "Analytics tracker.", &other.ID, json.RawMessage(`{"status":"done"}`))
}))
reloaded := loadCommonTrackerPattern(t, ctx, client, cp.ID)
@@ -218,20 +221,23 @@ func TestCommonTrackerPattern_Upsert_RequeuesBlankRowOnThirdPartyLink(t *testing
party := seedCommonThirdParty(t, ctx, client)
now := time.Now().UTC().Truncate(time.Microsecond)
enrichedAt := now.Add(-time.Hour)
attemptAt := now.Add(-time.Hour)
pattern := "requeue_" + gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType).String()
// Stage a terminal blank row: enriched, no description, no vendor.
// Stage a completed blank row: enriched (carries a payload), no
// description, no vendor, with prior attempts spent.
blank := coredata.CommonTrackerPattern{
ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType),
TrackerType: coredata.TrackerTypeCookie,
Pattern: pattern,
MatchType: coredata.TrackerPatternMatchTypeExact,
Description: "",
Confidence: 0.5,
EnrichedAt: &enrichedAt,
CreatedAt: now,
UpdatedAt: now,
ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType),
TrackerType: coredata.TrackerTypeCookie,
Pattern: pattern,
MatchType: coredata.TrackerPatternMatchTypeExact,
Description: "",
Confidence: 0.5,
Enrichment: json.RawMessage(`{"status":"no_result"}`),
EnrichmentAttempts: 2,
LastEnrichmentAttemptAt: &attemptAt,
CreatedAt: now,
UpdatedAt: now,
}
insertCommonTrackerPattern(t, ctx, client, blank)
@@ -263,7 +269,7 @@ func TestCommonTrackerPattern_Upsert_RequeuesBlankRowOnThirdPartyLink(t *testing
require.NotNil(t, reloaded.CommonThirdPartyID)
assert.Equal(t, party.ID, *reloaded.CommonThirdPartyID, "blank row must gain the linked third party")
assert.NotNil(t, reloaded.EnrichmentRequestedAt, "linking a vendor must re-queue the blank row for enrichment")
assert.Nil(t, reloaded.EnrichedAt, "re-queued row must no longer be terminal")
assert.Equal(t, 0, reloaded.EnrichmentAttempts, "re-queued row must get a fresh retry budget")
}
// TestCommonTrackerPattern_Upsert_KeepsDescribedRowTerminal pins the
@@ -279,19 +285,21 @@ func TestCommonTrackerPattern_Upsert_KeepsDescribedRowTerminal(t *testing.T) {
party := seedCommonThirdParty(t, ctx, client)
now := time.Now().UTC().Truncate(time.Microsecond)
enrichedAt := now.Add(-time.Hour)
attemptAt := now.Add(-time.Hour)
pattern := "described_" + gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType).String()
described := coredata.CommonTrackerPattern{
ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType),
TrackerType: coredata.TrackerTypeCookie,
Pattern: pattern,
MatchType: coredata.TrackerPatternMatchTypeExact,
Description: "An established analytics cookie.",
Confidence: 0.9,
EnrichedAt: &enrichedAt,
CreatedAt: now,
UpdatedAt: now,
ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType),
TrackerType: coredata.TrackerTypeCookie,
Pattern: pattern,
MatchType: coredata.TrackerPatternMatchTypeExact,
Description: "An established analytics cookie.",
Confidence: 0.9,
Enrichment: json.RawMessage(`{"status":"done"}`),
EnrichmentAttempts: 1,
LastEnrichmentAttemptAt: &attemptAt,
CreatedAt: now,
UpdatedAt: now,
}
insertCommonTrackerPattern(t, ctx, client, described)
@@ -315,5 +323,93 @@ func TestCommonTrackerPattern_Upsert_KeepsDescribedRowTerminal(t *testing.T) {
reloaded := loadCommonTrackerPattern(t, ctx, client, described.ID)
assert.Equal(t, "An established analytics cookie.", reloaded.Description, "existing description must be preserved")
assert.Nil(t, reloaded.EnrichmentRequestedAt, "described row must not be re-queued")
assert.NotNil(t, reloaded.EnrichedAt, "described row must stay terminal")
assert.NotEmpty(t, reloaded.Enrichment, "described row must keep its enrichment payload")
assert.Equal(t, 1, reloaded.EnrichmentAttempts, "described row's retry budget must be untouched")
}
// TestCommonTrackerPattern_ClearEnrichmentRequestedAt_CountsAttempt pins
// the claim contract: dequeuing a row counts the attempt and stamps the
// idle clock, so a crash before UpdateEnrichment still spends part of the
// retry budget and the stale clock starts at claim time.
func TestCommonTrackerPattern_ClearEnrichmentRequestedAt_CountsAttempt(t *testing.T) {
t.Parallel()
client := test.PGClient(t)
ctx := context.Background()
now := time.Now().UTC().Truncate(time.Microsecond)
requestedAt := now.Add(-time.Minute)
cp := coredata.CommonTrackerPattern{
ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType),
TrackerType: coredata.TrackerTypeCookie,
Pattern: "claim_" + gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType).String(),
MatchType: coredata.TrackerPatternMatchTypeExact,
Confidence: 0.5,
EnrichmentRequestedAt: &requestedAt,
CreatedAt: now,
UpdatedAt: now,
}
insertCommonTrackerPattern(t, ctx, client, cp)
require.NoError(t, client.WithTx(ctx, func(ctx context.Context, tx pg.Tx) error {
return cp.ClearEnrichmentRequestedAt(ctx, tx)
}))
reloaded := loadCommonTrackerPattern(t, ctx, client, cp.ID)
assert.Nil(t, reloaded.EnrichmentRequestedAt, "claim must remove the row from the queue")
assert.Equal(t, 1, reloaded.EnrichmentAttempts, "claim must count the attempt")
assert.NotNil(t, reloaded.LastEnrichmentAttemptAt, "claim must stamp the idle clock")
}
// TestCommonTrackerPattern_ResetStaleEnrichments_RespectsMaxAttempts pins
// the retry-budget contract: stale recovery re-queues a claimed-but-
// incomplete row that still has budget, but leaves a row at the
// max-attempts ceiling alone so a permanently failing row does not loop.
func TestCommonTrackerPattern_ResetStaleEnrichments_RespectsMaxAttempts(t *testing.T) {
t.Parallel()
client := test.PGClient(t)
ctx := context.Background()
now := time.Now().UTC().Truncate(time.Microsecond)
staleAttempt := now.Add(-time.Hour)
// A claimed-but-incomplete row with budget left: no payload yet, one
// attempt spent, idle past the threshold.
eligible := coredata.CommonTrackerPattern{
ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType),
TrackerType: coredata.TrackerTypeCookie,
Pattern: "stale_eligible_" + gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType).String(),
MatchType: coredata.TrackerPatternMatchTypeExact,
Confidence: 0.5,
EnrichmentAttempts: 1,
LastEnrichmentAttemptAt: &staleAttempt,
CreatedAt: now,
UpdatedAt: now,
}
insertCommonTrackerPattern(t, ctx, client, eligible)
// A row that has exhausted its retry budget must be left alone.
exhausted := coredata.CommonTrackerPattern{
ID: gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType),
TrackerType: coredata.TrackerTypeCookie,
Pattern: "stale_exhausted_" + gid.New(gid.NilTenant, coredata.CommonTrackerPatternEntityType).String(),
MatchType: coredata.TrackerPatternMatchTypeExact,
Confidence: 0.5,
EnrichmentAttempts: 3,
LastEnrichmentAttemptAt: &staleAttempt,
CreatedAt: now,
UpdatedAt: now,
}
insertCommonTrackerPattern(t, ctx, client, exhausted)
require.NoError(t, client.WithConn(ctx, func(ctx context.Context, conn pg.Querier) error {
return coredata.ResetStaleEnrichments(ctx, conn, time.Minute, 3)
}))
reloadedEligible := loadCommonTrackerPattern(t, ctx, client, eligible.ID)
assert.NotNil(t, reloadedEligible.EnrichmentRequestedAt, "stale row with budget must be re-queued")
reloadedExhausted := loadCommonTrackerPattern(t, ctx, client, exhausted.ID)
assert.Nil(t, reloadedExhausted.EnrichmentRequestedAt, "row at the max-attempts ceiling must not be re-queued")
}

37
pkg/coredata/migrations/20260615T161615Z.sql Normal file
View File

@@ -0,0 +1,37 @@
-- Copyright (c) 2026 Probo Inc <hello@probo.com>.
--
-- Permission to use, copy, modify, and/or distribute this software for any
-- purpose with or without fee is hereby granted, provided that the above
-- copyright notice and this permission notice appear in all copies.
--
-- THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
-- REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
-- AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
-- INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
-- LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
-- OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
-- PERFORMANCE OF THIS SOFTWARE.
-- Unify the enrichment-tracking model across common_tracker_patterns and
-- common_third_parties. common_tracker_patterns gains a JSONB provenance
-- payload and an attempt counter (mirroring common_third_parties), and its
-- enriched_at done-flag is renamed to last_enrichment_attempt_at, stamped at
-- claim time. "Enriched" is now detected via the enrichment payload presence,
-- not the timestamp.
ALTER TABLE common_tracker_patterns
ADD COLUMN enrichment JSONB,
ADD COLUMN enrichment_attempts INTEGER NOT NULL DEFAULT 0;
-- The DEFAULT only backfills existing rows; drop it so inserts must supply
-- the value explicitly.
ALTER TABLE common_tracker_patterns
ALTER COLUMN enrichment_attempts DROP DEFAULT;
ALTER TABLE common_tracker_patterns
RENAME COLUMN enriched_at TO last_enrichment_attempt_at;
-- common_third_parties already carries enrichment/enrichment_attempts; give
-- it the same explicit last-attempt timestamp (previously only recorded in
-- the enrichment JSON) so the stale-recovery clock has a dedicated column.
ALTER TABLE common_third_parties
ADD COLUMN last_enrichment_attempt_at TIMESTAMP WITH TIME ZONE;