repository

func (*EmbeddingsRepository) CountFeedbackRecordsForBackfillByInputKind ¶

func (r *EmbeddingsRepository) CountFeedbackRecordsForBackfillByInputKind(
	ctx context.Context,
	model string,
	inputKind models.EmbeddingInputKind,
) (int, error)

CountFeedbackRecordsForBackfillByInputKind counts globally eligible records missing model.

func (*EmbeddingsRepository) CountTenantFeedbackRecordsForBackfillByInputKind ¶

func (r *EmbeddingsRepository) CountTenantFeedbackRecordsForBackfillByInputKind(
	ctx context.Context,
	tenantID string,
	model string,
	inputKind models.EmbeddingInputKind,
) (int, error)

CountTenantFeedbackRecordsForBackfillByInputKind counts one tenant's eligible records missing model.

func (*EmbeddingsRepository) DeleteByFeedbackRecordAndModel ¶

func (r *EmbeddingsRepository) DeleteByFeedbackRecordAndModel(
	ctx context.Context, feedbackRecordID uuid.UUID, model string,
	stillCurrent func(fieldLabel, valueText, valueTextTranslated *string) bool,
) error

DeleteByFeedbackRecordAndModel removes the embedding row for the given feedback record and model. stillCurrent (optional) has the same stale-write guard semantics as Upsert: a clear enqueued for since-changed content must not delete the vector a newer job wrote.

func (*EmbeddingsRepository) DeleteEmbeddingsForOtherModels ¶

func (r *EmbeddingsRepository) DeleteEmbeddingsForOtherModels(
	ctx context.Context, currentModel string, batchSize int, additionalCurrentModels ...string,
) (int64, error)

DeleteEmbeddingsForOtherModels batch-deletes embedding rows whose model is not in the current model set. Reads only ever join on active models, so such rows are dead weight. Batched (batchSize rows per DELETE) so a large prune never holds long row locks or produces one giant WAL burst. Returns the total deleted. Run only after a model migration's backfill has completed, or reads using that model go dark until the new model's vectors exist.

func (*EmbeddingsRepository) GetEmbeddingAndTenantByFeedbackRecordAndModel ¶

func (r *EmbeddingsRepository) GetEmbeddingAndTenantByFeedbackRecordAndModel(
	ctx context.Context, feedbackRecordID uuid.UUID, model string,
) ([]float32, string, error)

GetEmbeddingAndTenantByFeedbackRecordAndModel returns the stored embedding and its feedback record tenant. Used by record-level similar feedback so the source record determines the tenant boundary for the search. Returns ErrEmbeddingNotFound when no embedding exists for the current model.

func (*EmbeddingsRepository) GetEmbeddingByFeedbackRecordAndModel ¶

func (r *EmbeddingsRepository) GetEmbeddingByFeedbackRecordAndModel(
	ctx context.Context, feedbackRecordID uuid.UUID, model string,
) ([]float32, error)

GetEmbeddingByFeedbackRecordAndModel returns the stored embedding for the given feedback record and model. Returns ErrEmbeddingNotFound when no row exists (record not embedded yet).

func (*EmbeddingsRepository) IterativeScanDegraded ¶

func (r *EmbeddingsRepository) IterativeScanDegraded() bool

IterativeScanDegraded reports whether HNSW iterative_scan has been latched off after the server rejected it (pgvector < 0.8). While true, nearest-neighbor recall is capped at ef_search until the process restarts. Surfaced as a gauge so the silent degradation is alertable, not just a one-time log line.

func (*EmbeddingsRepository) ListFeedbackRecordIDsForBackfill ¶

func (r *EmbeddingsRepository) ListFeedbackRecordIDsForBackfill(
	ctx context.Context, model string, afterID uuid.UUID, limit int,
) ([]uuid.UUID, error)

ListFeedbackRecordIDsForBackfill returns one keyset page (fr.id > afterID, ordered by id, at most limit rows) of feedback-record IDs that have non-empty value_text and no row in embeddings for the given model (so they need an embedding for that model). Pass uuid.Nil as afterID for the first page.

func (*EmbeddingsRepository) ListFeedbackRecordIDsForBackfillByInputKind ¶

func (r *EmbeddingsRepository) ListFeedbackRecordIDsForBackfillByInputKind(
	ctx context.Context,
	model string,
	inputKind models.EmbeddingInputKind,
	afterID uuid.UUID,
	limit int,
) ([]uuid.UUID, error)

ListFeedbackRecordIDsForBackfillByInputKind returns feedback-record IDs missing an embedding for model and eligible for the requested embedding input kind.

func (*EmbeddingsRepository) ListTenantFeedbackRecordIDsForBackfillByInputKind ¶

func (r *EmbeddingsRepository) ListTenantFeedbackRecordIDsForBackfillByInputKind(
	ctx context.Context,
	tenantID string,
	model string,
	inputKind models.EmbeddingInputKind,
	afterID uuid.UUID,
	limit int,
) ([]uuid.UUID, error)

ListTenantFeedbackRecordIDsForBackfillByInputKind returns one tenant's feedback-record IDs that are missing an embedding for model and eligible for the requested input kind.

func (*EmbeddingsRepository) NearestFeedbackRecordsByEmbedding ¶

func (r *EmbeddingsRepository) NearestFeedbackRecordsByEmbedding(
	ctx context.Context, model string, queryEmbedding []float32, tenantID string, limit int, excludeID *uuid.UUID, minScore float64,
) ([]models.FeedbackRecordWithScore, bool, error)

NearestFeedbackRecordsByEmbedding returns feedback record IDs and similarity scores (0..1) for the nearest neighbors to queryEmbedding, filtered by model and tenant. Rows with score < minScore are filtered in application code (not in WHERE) so pgvector's iterative index scan can run. The query vector is sent full-precision and implicitly cast to halfvec by the <=> operator (that cast is what makes the halfvec index usable). Sets hnsw.ef_search and iterative scan for recall. Over-fetches then trims to limit to account for tenant/minScore filtering. excludeID optionally excludes one feedback record (e.g. for "similar" endpoint). First page only; use NearestFeedbackRecordsByEmbeddingAfterCursor for next pages.

func (*EmbeddingsRepository) NearestFeedbackRecordsByEmbeddingAfterCursor ¶

func (r *EmbeddingsRepository) NearestFeedbackRecordsByEmbeddingAfterCursor(
	ctx context.Context, model string, queryEmbedding []float32, tenantID string, limit int,
	lastDistance float64, lastFeedbackRecordID uuid.UUID, excludeID *uuid.UUID, minScore float64,
) ([]models.FeedbackRecordWithScore, bool, error)

NearestFeedbackRecordsByEmbeddingAfterCursor returns the next page of nearest neighbors after the given cursor (lastDistance, lastFeedbackRecordID). Order is by (distance ASC, feedback_record_id ASC). minScore is applied in application code; query settings match NearestFeedbackRecordsByEmbedding. The cursor's lastDistance is the exact distance the previous page selected (not re-derived from the score), so the keyset comparison matches the stored ordering bit-for-bit.

func (*EmbeddingsRepository) TenantExistsForEmbeddingBackfill ¶

func (r *EmbeddingsRepository) TenantExistsForEmbeddingBackfill(ctx context.Context, tenantID string) (bool, error)

TenantExistsForEmbeddingBackfill reports whether Hub knows the tenant through feedback records or tenant settings. Hub has no authoritative tenants table, so checking both data sources lets the operator distinguish a likely typo from an existing tenant with no eligible missing embeddings.

func (*EmbeddingsRepository) Upsert ¶

func (r *EmbeddingsRepository) Upsert(
	ctx context.Context, feedbackRecordID uuid.UUID, model string, embedding []float32,
	stillCurrent func(fieldLabel, valueText, valueTextTranslated *string) bool,
) error

Upsert inserts or updates the embedding for (feedback_record_id, model). On conflict updates embedding and updated_at. Uses halfvec storage (2 bytes per dimension); pgvector-go converts float32 to float16 when encoding. embedding must have length models.EmbeddingVectorDimensions (fixed 768).

stillCurrent (optional) guards against the concurrent-jobs race: two jobs for the same record run in parallel and the one that read OLDER content lands its write LAST, permanently attaching a stale vector (the missing-rows-only backfill can never repair it). Under a per-record advisory lock the record's current content is re-read and compared; a mismatch returns huberrors.ErrEmbeddingSuperseded — a benign skip, since the job holding the current content writes the row.

func (*EnrichmentBacklogLeader) Close ¶

func (l *EnrichmentBacklogLeader) Close(ctx context.Context)

Close relinquishes leadership so another replica can take over promptly instead of waiting for this process's session to time out. Safe to call when not the leader.

func (*EnrichmentBacklogLeader) CountIfLeader ¶

func (l *EnrichmentBacklogLeader) CountIfLeader(
	ctx context.Context, defaultLang, taxonomyEmbeddingModel string,
) (EnrichmentStatusCounts, bool, error)

CountIfLeader returns the cross-tenant counts when this process holds (or wins) leadership, and reports whether it did. Not being the leader is the normal steady state for all but one replica, so it is signalled by a false second return rather than an error.

func (*EnrichmentFailuresRepository) RecordFailure ¶

func (r *EnrichmentFailuresRepository) RecordFailure(ctx context.Context, failure models.EnrichmentFailure) error

RecordFailure persists one enrichment failure.

Two outcomes are benign skips rather than errors worth failing a job over, and they are reported SEPARATELY because the rest of the worker treats them differently:

• ErrTenantWriteConflict — the tenant write lock was refused, so a purge is running. Which purge matters: the offboarding one takes the whole tenant, but the records-scoped one spares records newer than its high-water mark, so this does NOT imply the record is going away. Elsewhere in this worker that error means "retry".
• ErrNotFound — the foreign key failed, so the record was deleted between the enrichment attempt and this write. Retrying that could only fail again.

markFailed swallows both today, but collapsing them into one type would leave a trap for the next caller that propagates instead of swallowing: a deleted record would read as a transient conflict and be retried forever.

func (*EnrichmentStatusRepository) CountEnrichmentBacklogAggregate ¶

func (r *EnrichmentStatusRepository) CountEnrichmentBacklogAggregate(
	ctx context.Context, defaultLang string,
) (EnrichmentStatusCounts, error)

CountEnrichmentBacklogAggregate returns eligible/done counts per enrichment summed across all tenants. defaultLang is the deployment translation fallback. The result carries no tenant dimension. Prefer CountEnrichmentBacklogAggregateIfLeader from the poller so only one replica runs the scan.

func (*EnrichmentStatusRepository) CountEnrichmentStatus ¶

func (r *EnrichmentStatusRepository) CountEnrichmentStatus(
	ctx context.Context, tenantID, defaultLang string,
) (EnrichmentStatusCounts, error)

CountEnrichmentStatus returns one tenant's eligible/done counts per enrichment. defaultLang is the deployment translation fallback ("" disables the fallback, so only tenants with their own target language have eligible translation records). Always scoped to the given tenant_id.

func (*EnrichmentStatusRepository) CountFailedRecordsAggregate ¶

func (r *EnrichmentStatusRepository) CountFailedRecordsAggregate(
	ctx context.Context,
) ([]FailedRecordCount, error)

CountFailedRecordsAggregate returns the cross-tenant failed-record counts per enrichment.

Translation's un-enriched test is deliberately weaker than the per-tenant endpoint's: that one compares against the tenant's EFFECTIVE target, which needs the settings join and the deployment default. A gauge summed across tenants has no single target to compare against, so it asks only whether the record has any translation at all. The consequence is that a record translated into a since-changed target counts as done here while the endpoint counts it as pending — acceptable for a deployment-wide gauge, and written down so the two are not mistaken for a bug when they disagree.

func (*FeedbackRecordsRepository) ClearEmotions ¶

func (r *FeedbackRecordsRepository) ClearEmotions(
	ctx context.Context, feedbackRecordID uuid.UUID, stillCurrent func(valueText *string) bool,
) error

ClearEmotions removes a record's emotion enrichment AND its completion marker, returning it to the "not classified" state. Used when the source content is gone (an empty-content job clears rather than classifies), so the record is correctly excluded from progress counts instead of masquerading as classified-with-no-emotions.

func (*FeedbackRecordsRepository) Count ¶

func (r *FeedbackRecordsRepository) Count(
	ctx context.Context, filters *models.ListFeedbackRecordsFilters,
) (int, error)

Count returns the number of feedback records matching the given filters.

func (*FeedbackRecordsRepository) Create ¶

func (r *FeedbackRecordsRepository) Create(ctx context.Context, req *models.CreateFeedbackRecordRequest) (*models.FeedbackRecord, error)

Create inserts a new feedback record.

func (*FeedbackRecordsRepository) Delete ¶

func (r *FeedbackRecordsRepository) Delete(ctx context.Context, id uuid.UUID) error

Delete removes a feedback record.

func (*FeedbackRecordsRepository) DeleteByUser ¶

func (r *FeedbackRecordsRepository) DeleteByUser(
	ctx context.Context, filters *models.DeleteFeedbackRecordsByUserFilters,
) ([]models.DeletedFeedbackRecordsByTenant, error)

DeleteByUser deletes all feedback records matching user_id. When tenant_id is provided, deletion is restricted to that tenant; otherwise all user records are deleted across tenants (documented GDPR/right-to-erasure exception). Every spanned tenant's write lock is acquired before deleting; if any tenant is under purge the whole request fails with a retryable conflict. The delete is scoped to the locked tenants, so records appearing in new tenants mid-transaction are never touched without a lock. Because the tenant set is snapshotted before locking, a record could be written into a new (unlocked) tenant for the same user after the snapshot; after deleting, the same transaction re-checks for any in-scope record still present and, if found, returns a retryable conflict (rolling the whole delete back) rather than reporting an incomplete erasure as success. Erasure is idempotent, so the caller's retry converges once writes for the subject have stopped. It returns deleted IDs grouped by tenant so callers can publish tenant-scoped side effects.

func (*FeedbackRecordsRepository) GetByID ¶

func (r *FeedbackRecordsRepository) GetByID(ctx context.Context, id uuid.UUID) (*models.FeedbackRecord, error)

GetByID retrieves a single feedback record by ID. Embedding is not selected (API/worker reads stay lean).

func (*FeedbackRecordsRepository) List ¶

func (r *FeedbackRecordsRepository) List(
	ctx context.Context, filters *models.ListFeedbackRecordsFilters,
) ([]models.FeedbackRecord, bool, error)

List retrieves feedback records with optional filters. Embedding is not selected (API reads stay lean). Fetches limit+1 as sentinel to determine hasMore; returns trimmed slice and hasMore.

func (*FeedbackRecordsRepository) ListAfterCursor ¶

func (r *FeedbackRecordsRepository) ListAfterCursor(
	ctx context.Context, filters *models.ListFeedbackRecordsFilters, cursorValue time.Time, cursorID uuid.UUID,
) ([]models.FeedbackRecord, bool, error)

ListAfterCursor retrieves feedback records after the given keyset cursor.

cursorValue is the previous page's last row read from the column the listing is sorted by, so the caller and this predicate must agree on the ordering — which they do, because both derive it from the same filters. The cursor represents the last row of the previous page. Fetches limit+1 as sentinel to determine hasMore; returns trimmed slice and hasMore.

func (*FeedbackRecordsRepository) ListEmotionsBackfillTargets ¶

func (r *FeedbackRecordsRepository) ListEmotionsBackfillTargets(
	ctx context.Context, afterID uuid.UUID, limit int,
) ([]uuid.UUID, error)

ListEmotionsBackfillTargets returns one keyset page of eligible text records whose emotions are not yet set. Pass uuid.Nil as afterID for the first page.

func (*FeedbackRecordsRepository) ListSentimentBackfillTargets ¶

func (r *FeedbackRecordsRepository) ListSentimentBackfillTargets(
	ctx context.Context, afterID uuid.UUID, limit int,
) ([]uuid.UUID, error)

ListSentimentBackfillTargets returns one keyset page (id > afterID, ordered by id, at most limit rows) of eligible text records whose sentiment is not yet set. Used by the one-off classify backfill. Pass uuid.Nil as afterID for the first page (every UUIDv7 id sorts after it).

func (*FeedbackRecordsRepository) ListTranslationBackfillTargets ¶

func (r *FeedbackRecordsRepository) ListTranslationBackfillTargets(
	ctx context.Context, afterID uuid.UUID, limit int, defaultLang string,
) ([]models.TranslationBackfillTarget, error)

ListTranslationBackfillTargets returns one keyset page (fr.id > afterID, ordered by id, at most limit rows) of feedback records across all tenants that need (re)translation. Used by the one-off global backfill command. defaultLang is the fallback target for tenants with no target_language of their own ("" disables the fallback). Pass uuid.Nil as afterID for the first page.

func (*FeedbackRecordsRepository) ListTranslationBackfillTargetsForTenant ¶

func (r *FeedbackRecordsRepository) ListTranslationBackfillTargetsForTenant(
	ctx context.Context, tenantID string, afterID uuid.UUID, limit int, defaultLang string,
) ([]models.TranslationBackfillTarget, error)

ListTranslationBackfillTargetsForTenant returns one keyset page (fr.id > afterID, ordered by id, at most limit rows) of a single tenant's records that need (re)translation. The tenant filter + pagination let the settings-triggered backfill worker stream a large tenant without materializing every target at once. Pass uuid.Nil as afterID for the first page (every UUIDv7 id sorts after it).

func (*FeedbackRecordsRepository) SetEmotions ¶

func (r *FeedbackRecordsRepository) SetEmotions(
	ctx context.Context, feedbackRecordID uuid.UUID, emotions []models.EmotionValue,
	stillCurrent func(valueText *string) bool,
) error

SetEmotions stores or clears the emotion labels for a feedback record. Like SetSentiment the write is tenant-write-locked (so it cannot race a tenant data purge), publishes no domain event (emotions is a derived enrichment), and takes the same optional stillCurrent content guard: a mismatch against the record's current value_text returns huberrors.ErrClassificationSuperseded instead of landing a stale label last (see guardValueTextCurrent); nil ⇒ unconditional write. Emotions are multi-label: an empty or nil slice clears the column (writes NULL), a non-empty slice replaces it. The caller validates label membership; the column CHECKs are the final guard. A missing record returns NotFound.

func (*FeedbackRecordsRepository) SetSentiment ¶

func (r *FeedbackRecordsRepository) SetSentiment(
	ctx context.Context, feedbackRecordID uuid.UUID, sentiment *models.SentimentValue, score *float64,
	stillCurrent func(valueText *string) bool,
) error

SetSentiment stores or clears the sentiment label and score for a feedback record. The write is scoped to the record's tenant via the shared tenant write lock (so it cannot race a tenant data purge) and does NOT publish a domain event: sentiment is a derived enrichment, not a record edit, and re-publishing would loop the enrichment pipeline. A missing record (deleted or purged between read and write) returns NotFound.

Sentiment has no per-tenant target that a settings change could invalidate (unlike SetTranslation's target guard), but it shares the content race: stillCurrent (optional) re-reads the record's value_text atomically with the write and compares it to the content the classification was computed from; a mismatch returns huberrors.ErrClassificationSuperseded so a job that read older text cannot land its label last over a newer job's write — a stale non-NULL label would escape the NULL-rows-only backfill forever (see guardValueTextCurrent). nil ⇒ unconditional write. Passing a nil sentiment clears both columns (e.g. when value_text was emptied); a non-nil sentiment sets both. The caller pairs label and score; the column CHECKs are the final guard.

func (*FeedbackRecordsRepository) SetTranslation ¶

func (r *FeedbackRecordsRepository) SetTranslation(
	ctx context.Context, feedbackRecordID uuid.UUID, translated *string, langKey, defaultLang string,
	stillCurrent func(valueText *string) bool,
) error

SetTranslation stores the translated text and the target locale it was produced in for a feedback record. The write is scoped to the record's tenant via the shared tenant write lock (so it cannot race a tenant data purge) and does NOT publish a domain event: translation is a derived enrichment, not a record edit, and re-publishing would loop the enrichment pipeline. A missing record (deleted or purged between read and write) returns NotFound. translated may be nil.

Setting a translation (translated != nil) is conditional: it lands only while langKey still equals the tenant's current EFFECTIVE target — its own target_language, or defaultLang (TRANSLATION_DEFAULT_LANGUAGE) when it has none — otherwise it returns huberrors.ErrTranslationSuperseded. This makes the write atomic w.r.t. a concurrent target change and immune to a stale settings-cache read, so an out-of-order stale-target job cannot clobber a newer translation.

stillCurrent (optional) additionally guards BOTH paths against content churn: the record's current value_text is re-read atomically with the write and compared to the content the translation (or clear) was computed from; a mismatch returns ErrTranslationSuperseded so a job that read older text cannot land its result last and clobber a newer job's write (see guardValueTextCurrent). nil ⇒ no content guard.

func (*FeedbackRecordsRepository) Update ¶

func (r *FeedbackRecordsRepository) Update(
	ctx context.Context, id uuid.UUID, req *models.UpdateFeedbackRecordRequest,
) (updated, previous *models.FeedbackRecord, err error)

Update updates an existing feedback record. Only value fields, metadata, language, and user_id can be updated. It returns both the updated row and the pre-update ("previous") row so the caller can compute the fields that ACTUALLY changed against state consistent with this write: the previous snapshot is read FOR UPDATE inside the same transaction as the write, so a concurrent Update cannot change the row between the read and the write and make the diff stale.

func (*TaxonomyRepository) CountNodeRecords ¶

func (r *TaxonomyRepository) CountNodeRecords(
	ctx context.Context,
	runID uuid.UUID,
	tenantID string,
) ([]models.TaxonomyNodeRecordCount, error)

CountNodeRecords returns the feedback-record count for every visible node in a taxonomy run. Each count is a subtree total: the number of DISTINCT feedback records assigned (through cluster membership) to the node or any of its visible descendants. So a branch reports the count across all of its subtopics and the root reports the run total. The run must belong to the tenant, otherwise a not-found error is returned.

The count is derived from a single recursive descendant-closure query with COUNT(DISTINCT ...), which stays correct even if the same cluster is referenced by more than one node — a record is never double counted within a subtree. Records attach only through live cluster memberships, and a membership is removed by cascade when its feedback record is deleted, so counts track live data.

func (*TaxonomyRepository) CountScopeInput ¶

func (r *TaxonomyRepository) CountScopeInput(
	ctx context.Context,
	scope models.TaxonomyScope,
	embeddingModel string,
) (int, int, *string, error)

CountScopeInput counts text records and embeddings for a taxonomy scope.

func (*TaxonomyRepository) CreateRunIfAvailable ¶

func (r *TaxonomyRepository) CreateRunIfAvailable(
	ctx context.Context,
	params CreateTaxonomyRunParams,
) (*models.TaxonomyRun, bool, error)

CreateRunIfAvailable creates a taxonomy run unless one is already pending or running.

func (*TaxonomyRepository) FailRunIfStale ¶

func (r *TaxonomyRepository) FailRunIfStale(
	ctx context.Context,
	runID uuid.UUID,
	tenantID string,
	cutoff time.Time,
	message string,
	errorCode models.TaxonomyRunFailureCode,
) (*time.Time, error)

FailRunIfStale marks a pending/running taxonomy run as failed only when its liveness timestamp is still older than cutoff at the moment of the update. The status and freshness checks are part of the tenant-locked UPDATE so a heartbeat that lands after candidate selection protects the run. Returns the database terminal timestamp when the run was failed and nil when it was refreshed, completed, or removed first.

func (*TaxonomyRepository) FailStuckRuns ¶

func (r *TaxonomyRepository) FailStuckRuns(
	ctx context.Context,
	olderThan time.Duration,
	message string,
	errorCode models.TaxonomyRunFailureCode,
) ([]ReapedTaxonomyRun, error)

FailStuckRuns marks taxonomy runs stuck in a non-terminal state (pending/running) past olderThan as failed. Runs are orphaned when the taxonomy service crashes mid-run or its terminal callback is lost; without this sweep they are polled forever in the UI and block regeneration.

Staleness is measured against updated_at, the run's liveness signal: the taxonomy service bumps it via Heartbeat while a generation is in flight (see Heartbeat), so olderThan is the maximum tolerated gap between heartbeats, not a ceiling on total run duration. Until heartbeats flow, updated_at holds the timestamp of the run's last state change, so the sweep degrades to a coarse "no progress since" safety net.

Candidates are selected up front (a scoped update needs each run's tenant), then failed one at a time through FailRunIfStale so every mutation takes the shared tenant write lock — the repository invariant that coordinates tenant-owned writes with tenant-data purges. Stuck runs are rare, so a per-run loop rather than a batched per-tenant update is sufficient. The final UPDATE re-checks both status and updated_at under the tenant lock, so a run that heartbeats, reaches a terminal state, or is removed between selection and update is skipped. It returns the runs failed and the first unexpected error, if any.

func (*TaxonomyRepository) GetActiveRun ¶

func (r *TaxonomyRepository) GetActiveRun(ctx context.Context, scope models.TaxonomyScope) (*models.TaxonomyRun, error)

GetActiveRun returns the active taxonomy run for a scope.

func (*TaxonomyRepository) GetRunForInternalService ¶

func (r *TaxonomyRepository) GetRunForInternalService(
	ctx context.Context,
	runID uuid.UUID,
) (*models.TaxonomyRun, error)

GetRunForInternalService returns run metadata for internal taxonomy service-token workflows.

func (*TaxonomyRepository) GetRunForTenant ¶

func (r *TaxonomyRepository) GetRunForTenant(
	ctx context.Context,
	runID uuid.UUID,
	tenantID string,
) (*models.TaxonomyRun, error)

GetRunForTenant returns a taxonomy run by ID scoped to a tenant.

func (*TaxonomyRepository) GetRunInput ¶

func (r *TaxonomyRepository) GetRunInput(
	ctx context.Context,
	runID uuid.UUID,
	tenantID string,
	embeddingModel string,
) (*models.TaxonomyRunInputResponse, error)

GetRunInput returns feedback records and embeddings for a taxonomy run, capped at MaxTaxonomyRunInputRows (most recent first).

func (*TaxonomyRepository) GetRunInputRecordIDs ¶

func (r *TaxonomyRepository) GetRunInputRecordIDs(
	ctx context.Context,
	runID uuid.UUID,
	tenantID string,
) ([]uuid.UUID, error)

GetRunInputRecordIDs returns the exact bounded record selection a taxonomy result must cover. It intentionally omits feedback text and embeddings so completion does not materialize the full run input a second time.

func (*TaxonomyRepository) GetTree ¶

func (r *TaxonomyRepository) GetTree(
	ctx context.Context,
	runID uuid.UUID,
	tenantID string,
) (*models.TaxonomyTreeResponse, error)

GetTree returns the visible taxonomy tree for a run.

func (*TaxonomyRepository) Heartbeat ¶

func (r *TaxonomyRepository) Heartbeat(
	ctx context.Context,
	runID uuid.UUID,
	tenantID string,
) error

Heartbeat bumps updated_at to NOW() for a run still in a non-terminal state (pending/running), keeping it out of the stuck-run reaper's reach for another timeout window. The taxonomy service calls this periodically while a generation is in flight; updated_at is the liveness signal the reaper (FailStuckRuns) checks against.

The update runs through the shared tenant write lock — the repository invariant that coordinates tenant-owned writes with tenant-data purges. A heartbeat that finds no pending/running row (the run finished, was purged, or its id is stale) is a benign no-op: there is nothing left to keep alive, so it succeeds silently rather than erroring. Callers resolve the run (and thus a 404 for an unknown id) before reaching here.

func (*TaxonomyRepository) ListFieldOptions ¶

func (r *TaxonomyRepository) ListFieldOptions(
	ctx context.Context,
	tenantID string,
	embeddingModel string,
) ([]models.TaxonomyFieldOption, error)

ListFieldOptions returns taxonomy-capable feedback fields for a tenant.

func (*TaxonomyRepository) ListNodeRecords ¶

func (r *TaxonomyRepository) ListNodeRecords(
	ctx context.Context,
	nodeID uuid.UUID,
	tenantID string,
	limit int,
) ([]models.FeedbackRecord, int, error)

ListNodeRecords returns feedback records assigned to a visible taxonomy node or descendants. The node must be visible and belong to the tenant, otherwise a not-found error is returned — the same contract as every other node-scoped operation.

func (*TaxonomyRepository) ListRuns ¶

func (r *TaxonomyRepository) ListRuns(
	ctx context.Context,
	filters models.ListTaxonomyRunsFilters,
) ([]models.TaxonomyRun, error)

ListRuns returns taxonomy run history for a tenant and optional scope filters.

func (*TaxonomyRepository) MarkRunFailed ¶

func (r *TaxonomyRepository) MarkRunFailed(
	ctx context.Context,
	runID uuid.UUID,
	tenantID string,
	message string,
	errorCode models.TaxonomyRunFailureCode,
	metrics json.RawMessage,
) (*models.TaxonomyRun, error)

MarkRunFailed transitions a taxonomy run to failed with an error message.

func (*TaxonomyRepository) MarkRunRunning ¶

func (r *TaxonomyRepository) MarkRunRunning(
	ctx context.Context,
	runID uuid.UUID,
	tenantID string,
) (*models.TaxonomyRun, error)

MarkRunRunning transitions a taxonomy run to running.

func (*TaxonomyRepository) RemoveNode ¶

func (r *TaxonomyRepository) RemoveNode(
	ctx context.Context,
	nodeID uuid.UUID,
	tenantID string,
	actorID string,
) (*models.TaxonomyNode, error)

RemoveNode soft-removes a taxonomy node and records an edit event.

func (*TaxonomyRepository) RenameNode ¶

func (r *TaxonomyRepository) RenameNode(
	ctx context.Context,
	nodeID uuid.UUID,
	tenantID string,
	actorID string,
	label string,
) (*models.TaxonomyNode, error)

RenameNode updates a taxonomy node label and records an edit event.

func (*TaxonomyRepository) StoreResultAndActivate ¶

func (r *TaxonomyRepository) StoreResultAndActivate(
	ctx context.Context,
	runID uuid.UUID,
	tenantID string,
	req models.TaxonomyRunResultRequest,
) (*models.TaxonomyRun, error)

StoreResultAndActivate stores generated taxonomy artifacts and activates the run.

func (*TenantDataRepository) DeleteByTenant ¶

func (r *TenantDataRepository) DeleteByTenant(ctx context.Context, tenantID string) (*models.TenantDataDeleteCounts, error)

DeleteByTenant deletes all Hub-owned data for a tenant and returns per-resource counts.

func (*TenantDataRepository) PurgeFeedbackRecordsByTenant ¶

func (r *TenantDataRepository) PurgeFeedbackRecordsByTenant(
	ctx context.Context, tenantID string,
) (*models.FeedbackRecordsPurgeCounts, error)

PurgeFeedbackRecordsByTenant deletes every feedback record for a tenant, everything derived from those records, and the taxonomy built on them, returning exact per-table counts. Unlike DeleteByTenant it leaves the tenant's *configuration* — webhooks and settings — intact, so the dataset stays usable and any integrator setup survives.

Two phases, because the two halves have different shapes. Records are unbounded, so they go in committed batches (see feedbackRecordsPurgeBatchSize): resumable, and the counts reported are the ones actually achieved even on a partial failure. Taxonomy artifacts are bounded by the number of runs, so once the records are gone they are removed in one final transaction.

Bounded to the records that existed when the purge started, via a high-water mark on the uuidv7 id. Two reasons, both load-bearing: the exclusive lock is released between batches, so without a bound a tenant that keeps ingesting keeps feeding the loop and the purge may never reach a zero batch; and an operator asked to empty what they saw, not to delete feedback that arrived afterwards.

Ordering matters: the taxonomy phase runs last so that a purge interrupted midway leaves records missing but the tree still standing, rather than the reverse — a tree with no records is the misleading state, and it is exactly what the final phase removes.

func (*TenantSettingsRepository) Get ¶

func (r *TenantSettingsRepository) Get(
	ctx context.Context, tenantID string,
) (*models.TenantSettings, bool, error)

Get returns the tenant's settings row and found=true, or (nil, false, nil) when the tenant has none. The query is always scoped to the given tenant_id, so one tenant's settings can never be read under another tenant.

func (*TenantSettingsRepository) Patch ¶

func (r *TenantSettingsRepository) Patch(
	ctx context.Context, tenantID string, set models.EnrichmentSettings, removeKeys []string,
) (*models.TenantSettings, error)

Patch applies an RFC 7396 JSON Merge Patch to the tenant's settings: keys in set are written (a top-level JSONB `||`), keys in removeKeys are deleted (`- text[]`), and keys mentioned in neither are left untouched. For a tenant with no row yet the set object becomes the initial settings. set and removeKeys are disjoint, so the merge-then-remove order does not matter.

func (*TenantSettingsRepository) Upsert ¶

func (r *TenantSettingsRepository) Upsert(
	ctx context.Context, tenantID string, settings models.EnrichmentSettings,
) (*models.TenantSettings, error)

Upsert creates or replaces (full replace) the tenant's settings and returns the stored row.

func (*WebhooksRepository) Count ¶

func (r *WebhooksRepository) Count(ctx context.Context, filters *models.ListWebhooksFilters) (int64, error)

Count returns the total count of webhooks matching the filters.

func (*WebhooksRepository) Create ¶

func (r *WebhooksRepository) Create(ctx context.Context, req *models.CreateWebhookRequest) (*models.Webhook, error)

Create inserts a new webhook.

func (*WebhooksRepository) Delete ¶

func (r *WebhooksRepository) Delete(ctx context.Context, id uuid.UUID) (*models.DeletedWebhook, error)

Delete removes a webhook and returns the deleted tenant boundary for side effects.

func (*WebhooksRepository) GetByID ¶

func (r *WebhooksRepository) GetByID(ctx context.Context, id uuid.UUID) (*models.Webhook, error)

GetByID retrieves a single webhook by ID.

func (*WebhooksRepository) List ¶

func (r *WebhooksRepository) List(ctx context.Context, filters *models.ListWebhooksFilters) ([]models.Webhook, bool, error)

List retrieves webhooks with optional filters. Fetches limit+1 as sentinel to determine hasMore; returns trimmed slice and hasMore.

func (*WebhooksRepository) ListAfterCursor ¶

func (r *WebhooksRepository) ListAfterCursor(
	ctx context.Context, filters *models.ListWebhooksFilters, cursorCreatedAt time.Time, cursorID uuid.UUID,
) ([]models.Webhook, bool, error)

ListAfterCursor retrieves webhooks after the given keyset cursor (created_at, id). Order is created_at DESC, id ASC. The cursor represents the last row of the previous page. Fetches limit+1 as sentinel to determine hasMore; returns trimmed slice and hasMore.

func (*WebhooksRepository) ListEnabledForEventTypeAndTenant ¶

func (r *WebhooksRepository) ListEnabledForEventTypeAndTenant(
	ctx context.Context, eventType string, tenantID *string,
) ([]models.Webhook, error)

ListEnabledForEventTypeAndTenant retrieves enabled webhooks for an event type and tenant boundary. Webhooks match only the same tenant. A missing tenantID matches nothing.

func (*WebhooksRepository) Update ¶

func (r *WebhooksRepository) Update(ctx context.Context, id uuid.UUID, req *models.UpdateWebhookRequest) (*models.Webhook, error)

Update updates an existing webhook.