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

Add tracker description enrichment worker

Browse Source 
Tracker descriptions were only filled on the agent-identification path,
so patterns resolved by domain, sibling, or fallback stayed without one,
and empty mapping upserts could clobber a researched description on the
shared catalog row.

Move description ownership to a dedicated, global common-pattern
enrichment worker. New catalog rows are queued on insert; the worker
researches a compliance-grade description with web search, records it on
the common pattern, and fans it out to every linked tracker pattern. The
mapping worker no longer generates descriptions and only propagates an
already-enriched one at link time.

Rename TrackerMappingConfig to TrackerAgentsConfig since the mapping and
enrichment agents now share it.

Signed-off-by: Émile Ré <emile@probo.com>
This commit is contained in:
Émile Ré
2026-05-29 11:48:10 +02:00
parent 29791ae775
commit 24bece6f86
12 changed files with 665 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

220
pkg/coredata/common_tracker_pattern.go
View File

@@ -27,16 +27,18 @@ 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"`
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"`
EnrichedAt *time.Time `db:"enriched_at"`
CreatedAt time.Time `db:"created_at"`
UpdatedAt time.Time `db:"updated_at"`
}
CommonTrackerPatterns []*CommonTrackerPattern
@@ -57,6 +59,8 @@ SELECT
description,
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
created_at,
updated_at
FROM
@@ -104,6 +108,8 @@ SELECT
description,
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
created_at,
updated_at
FROM
@@ -154,6 +160,8 @@ INSERT INTO common_tracker_patterns (
description,
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
created_at,
updated_at
) VALUES (
@@ -165,22 +173,26 @@ INSERT INTO common_tracker_patterns (
@description,
@max_age_seconds,
@confidence,
@enrichment_requested_at,
@enriched_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,
"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,
"enriched_at": p.EnrichedAt,
"created_at": p.CreatedAt,
"updated_at": p.UpdatedAt,
}
_, err := conn.Exec(ctx, q, args)
@@ -195,6 +207,12 @@ func (p *CommonTrackerPattern) Upsert(
ctx context.Context,
conn pg.Tx,
) (inserted bool, err error) {
// On insert, a description-less row is immediately queued for the
// enrichment worker (enrichment_requested_at = NOW()). On conflict the
// enrichment columns are left untouched, and an empty incoming
// description never overwrites an existing one — descriptions are owned
// by the enrichment worker, so mapping-side upserts must not clobber a
// researched description with an empty string.
q := `
INSERT INTO common_tracker_patterns (
id,
@@ -205,6 +223,8 @@ INSERT INTO common_tracker_patterns (
description,
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
created_at,
updated_at
) VALUES (
@@ -216,6 +236,8 @@ INSERT INTO common_tracker_patterns (
@description,
@max_age_seconds,
@confidence,
CASE WHEN @description = '' THEN NOW() ELSE NULL END,
NULL,
@created_at,
@updated_at
)
@@ -223,7 +245,10 @@ ON CONFLICT (tracker_type, pattern, COALESCE(max_age_seconds, -1)) DO UPDATE
SET
common_third_party_id = EXCLUDED.common_third_party_id,
match_type = EXCLUDED.match_type,
description = EXCLUDED.description,
description = CASE
WHEN EXCLUDED.description = '' THEN common_tracker_patterns.description
ELSE EXCLUDED.description
END,
confidence = EXCLUDED.confidence,
updated_at = EXCLUDED.updated_at
RETURNING
@@ -235,6 +260,8 @@ RETURNING
description,
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
created_at,
updated_at
`
@@ -303,6 +330,8 @@ SELECT
description,
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
created_at,
updated_at
FROM
@@ -418,6 +447,8 @@ SELECT
description,
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
created_at,
updated_at
FROM
@@ -444,6 +475,151 @@ ORDER BY pattern ASC;
return nil
}
// LoadNextForEnrichmentForUpdateSkipLocked claims the next common tracker
// pattern queued for description enrichment, oldest request first. It
// mirrors the mapping worker's claim pattern: the row is locked FOR
// UPDATE SKIP LOCKED so concurrent enrichment workers never pick the same
// row.
func (p *CommonTrackerPattern) LoadNextForEnrichmentForUpdateSkipLocked(
ctx context.Context,
tx pg.Tx,
) error {
q := `
SELECT
id,
common_third_party_id,
tracker_type,
pattern,
match_type,
description,
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
created_at,
updated_at
FROM
common_tracker_patterns
WHERE
enrichment_requested_at IS NOT NULL
ORDER BY
enrichment_requested_at ASC
FOR UPDATE SKIP LOCKED
LIMIT 1;
`
rows, err := tx.Query(ctx, q)
if err != nil {
return fmt.Errorf("cannot query common tracker pattern for enrichment: %w", err)
}
pattern, err := pgx.CollectExactlyOneRow(rows, pgx.RowToStructByName[CommonTrackerPattern])
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return ErrResourceNotFound
}
return fmt.Errorf("cannot collect common tracker pattern for enrichment: %w", err)
}
*p = pattern
return nil
}
// ClearEnrichmentRequestedAt removes the row from the enrichment queue. It
// bumps updated_at so the stale-recovery clock starts at claim time.
func (p *CommonTrackerPattern) ClearEnrichmentRequestedAt(
ctx context.Context,
tx pg.Tx,
) error {
q := `
UPDATE common_tracker_patterns
SET
enrichment_requested_at = NULL,
updated_at = NOW()
WHERE id = @id
`
args := pgx.StrictNamedArgs{"id": p.ID}
_, err := tx.Exec(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot clear enrichment requested at: %w", err)
}
p.EnrichmentRequestedAt = nil
return nil
}
// SetEnriched records the researched description and marks the row
// terminally enriched so it is never re-queued.
func (p *CommonTrackerPattern) SetEnriched(
ctx context.Context,
tx pg.Tx,
description string,
) error {
q := `
UPDATE common_tracker_patterns
SET
description = @description,
enriched_at = NOW(),
enrichment_requested_at = NULL,
updated_at = NOW()
WHERE id = @id
`
args := pgx.StrictNamedArgs{
"id": p.ID,
"description": description,
}
result, err := tx.Exec(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot mark common tracker pattern enriched: %w", err)
}
if result.RowsAffected() == 0 {
return ErrResourceNotFound
}
p.Description = description
return nil
}
// 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.
func ResetStaleEnrichments(
ctx context.Context,
conn pg.Querier,
staleAfter time.Duration,
) error {
q := `
UPDATE common_tracker_patterns
SET
enrichment_requested_at = NOW(),
updated_at = NOW()
WHERE
enrichment_requested_at IS NULL
AND enriched_at IS NULL
AND description = ''
AND updated_at < @stale_before
`
args := pgx.StrictNamedArgs{"stale_before": time.Now().Add(-staleAfter)}
_, err := conn.Exec(ctx, q, args)
if err != nil {
return fmt.Errorf("cannot reset stale common tracker pattern enrichments: %w", err)
}
return nil
}
func (ps *CommonTrackerPatterns) LoadByIDs(
ctx context.Context,
conn pg.Querier,
@@ -459,6 +635,8 @@ SELECT
description,
max_age_seconds,
confidence,
enrichment_requested_at,
enriched_at,
created_at,
updated_at
FROM

31
pkg/coredata/migrations/20260529T091300Z.sql Normal file
View File

@@ -0,0 +1,31 @@
-- Copyright (c) 2026 Probo Inc <hello@getprobo.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.
-- The common-pattern enrichment worker fills descriptions on
-- common_tracker_patterns using an agent with web search, then fans the
-- result out to every linked tracker pattern. enrichment_requested_at is
-- the work queue (claimed FOR UPDATE SKIP LOCKED); enriched_at marks a
-- row as terminally enriched so it is never re-enqueued.
ALTER TABLE common_tracker_patterns
ADD COLUMN enrichment_requested_at TIMESTAMP WITH TIME ZONE,
ADD COLUMN enriched_at TIMESTAMP WITH TIME ZONE;
CREATE INDEX idx_common_tracker_patterns_enrichment
ON common_tracker_patterns (enrichment_requested_at)
WHERE enrichment_requested_at IS NOT NULL;
-- Enqueue existing description-less rows for a first enrichment pass.
UPDATE common_tracker_patterns
SET enrichment_requested_at = NOW()
WHERE description = '';

37
pkg/coredata/tracker_pattern.go
View File

@@ -1303,3 +1303,40 @@ WHERE
return result.RowsAffected(), nil
}
// BackfillDescriptionByCommonTrackerPatternID copies an enriched
// description onto every tracker pattern linked to the given common
// pattern that does not yet have one. It is invoked by the common-pattern
// enrichment worker, a global system process, so it is intentionally not
// tenant-scoped: a single catalog enrichment fans out to all tenants'
// linked patterns. The description = ” guard guarantees a pattern that
// already carries a description is never overwritten. Returns the number
// of patterns backfilled.
func (tps *TrackerPatterns) BackfillDescriptionByCommonTrackerPatternID(
ctx context.Context,
tx pg.Tx,
commonTrackerPatternID gid.GID,
description string,
) (int64, error) {
q := `
UPDATE tracker_patterns
SET
description = @description,
updated_at = NOW()
WHERE
common_tracker_pattern_id = @common_tracker_pattern_id
AND description = ''
`
args := pgx.StrictNamedArgs{
"common_tracker_pattern_id": commonTrackerPatternID,
"description": description,
}
result, err := tx.Exec(ctx, q, args)
if err != nil {
return 0, fmt.Errorf("cannot backfill tracker pattern descriptions: %w", err)
}
return result.RowsAffected(), nil
}