Amirhossein Khalili
5673b60ba6
Bootstrap independently reviewed catalogs through audited import, approval, and initial activation transitions, then resume insight scheduling safely.
232 lines
11 KiB
Go
232 lines
11 KiB
Go
package eventcatalog
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
type Import struct {
|
|
Version string `json:"version"`
|
|
CoverageStartJalaliYear int `json:"coverageStartJalaliYear"`
|
|
CoverageEndJalaliYear int `json:"coverageEndJalaliYear"`
|
|
SourceURI string `json:"sourceUri"`
|
|
SourceVersion string `json:"sourceVersion"`
|
|
SourceChecksum string `json:"sourceChecksum"`
|
|
ImportedBy string `json:"importedBy"`
|
|
Events []Event `json:"events"`
|
|
}
|
|
|
|
func ValidateImport(input Import, currentJalaliYear int) error {
|
|
if strings.TrimSpace(input.Version) == "" || strings.TrimSpace(input.SourceURI) == "" || strings.TrimSpace(input.SourceVersion) == "" || strings.TrimSpace(input.SourceChecksum) == "" || strings.TrimSpace(input.ImportedBy) == "" {
|
|
return errors.New("event catalog: import metadata is incomplete")
|
|
}
|
|
if input.CoverageStartJalaliYear > currentJalaliYear-3 || input.CoverageEndJalaliYear < currentJalaliYear+2 {
|
|
return fmt.Errorf("event catalog: coverage must include Jalali years %d through %d", currentJalaliYear-3, currentJalaliYear+2)
|
|
}
|
|
if len(input.Events) == 0 {
|
|
return errors.New("event catalog: import contains no events")
|
|
}
|
|
if err := ValidateEvents(input.Events); err != nil {
|
|
return err
|
|
}
|
|
return ValidateDeclaredCoverage(input.Events, input.CoverageStartJalaliYear, input.CoverageEndJalaliYear)
|
|
}
|
|
|
|
func ValidateApproval(importedBy, reviewedBy string) error {
|
|
if strings.TrimSpace(reviewedBy) == "" {
|
|
return errors.New("event catalog: reviewer is required")
|
|
}
|
|
if strings.EqualFold(strings.TrimSpace(importedBy), strings.TrimSpace(reviewedBy)) {
|
|
return errors.New("event catalog: importer cannot approve the same catalog")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func ValidateReviewableChecksum(checksum string) error {
|
|
normalized := strings.ToLower(strings.TrimSpace(checksum))
|
|
if normalized == "" || strings.Contains(normalized, "review-required") || strings.Contains(normalized, "pending-review") || strings.Contains(normalized, "placeholder") {
|
|
return errors.New("event catalog: source checksum is not independently reviewed")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type Repository struct{ pool *pgxpool.Pool }
|
|
|
|
func NewRepository(pool *pgxpool.Pool) *Repository { return &Repository{pool: pool} }
|
|
|
|
func (r *Repository) CatalogState(ctx context.Context, version string) (CatalogState, error) {
|
|
return r.catalogState(ctx, `SELECT version,status,coverage_start_jalali_year,coverage_end_jalali_year,source_uri,source_version,source_checksum,imported_by FROM calendar_catalog_versions WHERE version=$1`, version)
|
|
}
|
|
|
|
func (r *Repository) ActiveCatalogState(ctx context.Context) (CatalogState, error) {
|
|
return r.catalogState(ctx, `SELECT version,status,coverage_start_jalali_year,coverage_end_jalali_year,source_uri,source_version,source_checksum,imported_by FROM calendar_catalog_versions WHERE status='ACTIVE'`)
|
|
}
|
|
|
|
func (r *Repository) catalogState(ctx context.Context, query string, arguments ...any) (CatalogState, error) {
|
|
var state CatalogState
|
|
err := r.pool.QueryRow(ctx, query, arguments...).Scan(&state.Version, &state.Status, &state.CoverageStartJalaliYear, &state.CoverageEndJalaliYear, &state.SourceURI, &state.SourceVersion, &state.SourceChecksum, &state.ImportedBy)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return CatalogState{}, ErrCatalogNotFound
|
|
}
|
|
if err != nil {
|
|
return CatalogState{}, fmt.Errorf("querying calendar catalog state: %w", err)
|
|
}
|
|
return state, nil
|
|
}
|
|
|
|
func (r *Repository) ImportDraft(ctx context.Context, input Import) error {
|
|
if err := ValidateEvents(input.Events); err != nil {
|
|
return err
|
|
}
|
|
tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{})
|
|
if err != nil {
|
|
return fmt.Errorf("starting calendar import: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
|
|
var versionID int64
|
|
err = tx.QueryRow(ctx, `INSERT INTO calendar_catalog_versions(version,status,coverage_start_jalali_year,coverage_end_jalali_year,source_uri,source_version,source_checksum,imported_by)
|
|
VALUES($1,'DRAFT',$2,$3,$4,$5,$6,$7) RETURNING id`, input.Version, input.CoverageStartJalaliYear, input.CoverageEndJalaliYear, input.SourceURI, input.SourceVersion, input.SourceChecksum, input.ImportedBy).Scan(&versionID)
|
|
if err != nil {
|
|
return fmt.Errorf("creating calendar catalog version: %w", err)
|
|
}
|
|
for _, event := range input.Events {
|
|
if event.Variant == "" {
|
|
event.Variant = "PRIMARY"
|
|
}
|
|
categories, err := json.Marshal(event.MerchantCategories)
|
|
if err != nil {
|
|
return fmt.Errorf("encoding calendar event categories: %w", err)
|
|
}
|
|
channels, err := json.Marshal(event.BehavioralChannels)
|
|
if err != nil {
|
|
return fmt.Errorf("encoding calendar event channels: %w", err)
|
|
}
|
|
aliases, err := json.Marshal(event.Aliases)
|
|
if err != nil {
|
|
return fmt.Errorf("encoding calendar event aliases: %w", err)
|
|
}
|
|
themes, err := json.Marshal(event.Themes)
|
|
if err != nil {
|
|
return fmt.Errorf("encoding calendar event themes: %w", err)
|
|
}
|
|
tags, err := json.Marshal(event.CampaignTags)
|
|
if err != nil {
|
|
return fmt.Errorf("encoding calendar event campaign tags: %w", err)
|
|
}
|
|
_, err = tx.Exec(ctx, `INSERT INTO calendar_events(catalog_version_id,event_key,occasion_key,event_variant,title_fa,title_en,aliases,themes,campaign_tags,start_date,end_date,recurrence_rule,event_type,tone,commercial_posture,official_holiday,source_uri,source_version,source_checksum,inclusion_rationale,geographic_scope,merchant_categories,behavioral_channels,lead_days,lag_days,relevance_bps,confidence_bps,importance_bps)
|
|
VALUES($1,$2,$3,$4,$5,nullif($6,''),$7,$8,$9,$10,$11,nullif($12,''),$13,$14,$15,$16,$17,$18,nullif($19,''),$20,$21,$22,$23,$24,$25,$26,$27,$28)`, versionID, event.ID, event.OccasionKey, event.Variant, event.TitleFA, event.TitleEN, aliases, themes, tags, event.StartDate, event.EndDate, event.Recurrence, event.Type, event.Tone, event.CommercialPosture, event.OfficialHoliday, event.SourceURI, event.SourceVersion, event.SourceChecksum, event.InclusionRationale, event.GeographicScope, categories, channels, event.LeadDays, event.LagDays, event.RelevanceBPS, event.ConfidenceBPS, event.ImportanceBPS)
|
|
if err != nil {
|
|
return fmt.Errorf("inserting calendar event %q: %w", event.ID, err)
|
|
}
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO calendar_catalog_audit(catalog_version_id,action,actor) VALUES($1,'IMPORTED',$2)`, versionID, input.ImportedBy); err != nil {
|
|
return fmt.Errorf("auditing calendar import: %w", err)
|
|
}
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return fmt.Errorf("committing calendar import: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) Approve(ctx context.Context, version, reviewer string) error {
|
|
tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{})
|
|
if err != nil {
|
|
return fmt.Errorf("starting calendar approval: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
|
|
var id int64
|
|
var importedBy string
|
|
var sourceChecksum string
|
|
if err := tx.QueryRow(ctx, `SELECT id,imported_by,source_checksum FROM calendar_catalog_versions WHERE version=$1 AND status='DRAFT' FOR UPDATE`, version).Scan(&id, &importedBy, &sourceChecksum); err != nil {
|
|
return fmt.Errorf("loading draft calendar catalog: %w", err)
|
|
}
|
|
if err := ValidateApproval(importedBy, reviewer); err != nil {
|
|
return err
|
|
}
|
|
if err := ValidateReviewableChecksum(sourceChecksum); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE calendar_catalog_versions SET status='APPROVED',reviewed_by=$2,reviewed_at=now() WHERE id=$1`, id, reviewer); err != nil {
|
|
return fmt.Errorf("approving calendar catalog: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO calendar_catalog_audit(catalog_version_id,action,actor) VALUES($1,'APPROVED',$2)`, id, reviewer); err != nil {
|
|
return fmt.Errorf("auditing calendar approval: %w", err)
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
func (r *Repository) Activate(ctx context.Context, version, actor string) error {
|
|
tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{})
|
|
if err != nil {
|
|
return fmt.Errorf("starting calendar activation: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
|
|
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended('calendar-catalog-activation',0))`); err != nil {
|
|
return fmt.Errorf("locking calendar activation: %w", err)
|
|
}
|
|
var id int64
|
|
var sourceChecksum string
|
|
if err := tx.QueryRow(ctx, `SELECT id,source_checksum FROM calendar_catalog_versions WHERE version=$1 AND status='APPROVED' FOR UPDATE`, version).Scan(&id, &sourceChecksum); err != nil {
|
|
return fmt.Errorf("loading approved calendar catalog: %w", err)
|
|
}
|
|
if err := ValidateReviewableChecksum(sourceChecksum); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE calendar_catalog_versions SET status='RETIRED' WHERE status='ACTIVE'`); err != nil {
|
|
return fmt.Errorf("retiring active calendar catalog: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE calendar_catalog_versions SET status='ACTIVE',activated_at=now() WHERE id=$1`, id); err != nil {
|
|
return fmt.Errorf("activating calendar catalog: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO calendar_catalog_audit(catalog_version_id,action,actor) VALUES($1,'ACTIVATED',$2)`, id, actor); err != nil {
|
|
return fmt.Errorf("auditing calendar activation: %w", err)
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
func (r *Repository) ActivateInitial(ctx context.Context, version, actor string) error {
|
|
if strings.TrimSpace(actor) == "" {
|
|
return errors.New("event catalog: activation operator is required")
|
|
}
|
|
tx, err := r.pool.BeginTx(ctx, pgx.TxOptions{})
|
|
if err != nil {
|
|
return fmt.Errorf("starting initial calendar activation: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
|
|
if _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended('calendar-catalog-activation',0))`); err != nil {
|
|
return fmt.Errorf("locking initial calendar activation: %w", err)
|
|
}
|
|
var activeVersion string
|
|
err = tx.QueryRow(ctx, `SELECT version FROM calendar_catalog_versions WHERE status='ACTIVE'`).Scan(&activeVersion)
|
|
if err == nil {
|
|
if activeVersion == version {
|
|
return tx.Commit(ctx)
|
|
}
|
|
return fmt.Errorf("%w: active version %q differs from requested version %q", ErrCatalogConflict, activeVersion, version)
|
|
}
|
|
if !errors.Is(err, pgx.ErrNoRows) {
|
|
return fmt.Errorf("loading active calendar catalog: %w", err)
|
|
}
|
|
var id int64
|
|
var sourceChecksum string
|
|
if err := tx.QueryRow(ctx, `SELECT id,source_checksum FROM calendar_catalog_versions WHERE version=$1 AND status='APPROVED' FOR UPDATE`, version).Scan(&id, &sourceChecksum); err != nil {
|
|
return fmt.Errorf("loading approved calendar catalog: %w", err)
|
|
}
|
|
if err := ValidateReviewableChecksum(sourceChecksum); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(ctx, `UPDATE calendar_catalog_versions SET status='ACTIVE',activated_at=now() WHERE id=$1`, id); err != nil {
|
|
return fmt.Errorf("activating calendar catalog: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `INSERT INTO calendar_catalog_audit(catalog_version_id,action,actor) VALUES($1,'ACTIVATED',$2)`, id, actor); err != nil {
|
|
return fmt.Errorf("auditing calendar activation: %w", err)
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|