package db import ( "context" "database/sql" ) // UpsertCheck records (or, on retry, overwrites) a single check result. // registry_id/cycle_id are resolved from c.IPID's current ip_queue row at // write time, so callers (agentcore/probercore) never need to know about // the registry — see migrations/0007_ip_registry.sql. The // UNIQUE(registry_id, cycle_id, source, check_type, target) constraint plus // this upsert is what makes agent/prober result submission safely // retryable without producing duplicate rows, now scoped to the durable // per-address cycle rather than the ip_queue row's attempt_number, so it // survives that row being deleted and the address later resubmitted. func (d *DB) UpsertCheck(ctx context.Context, c Check) error { _, err := d.ExecContext(ctx, ` INSERT INTO checks (registry_id, cycle_id, ip_id, ip_address, attempt_number, validator_id, source, check_type, target, success, latency_ms, detail, checked_at, created_at) VALUES ((SELECT registry_id FROM ip_queue WHERE id=?), (SELECT cycle_id FROM ip_queue WHERE id=?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(registry_id, cycle_id, source, check_type, target) DO UPDATE SET validator_id=excluded.validator_id, success=excluded.success, latency_ms=excluded.latency_ms, detail=excluded.detail, checked_at=excluded.checked_at `, c.IPID, c.IPID, c.IPID, c.IPAddress, c.AttemptNumber, c.ValidatorID, c.Source, c.CheckType, c.Target, c.Success, c.LatencyMS, c.Detail, timeToDB(c.CheckedAt), timeToDB(Now())) return err } const checksSelect = ` SELECT id, registry_id, cycle_id, ip_id, ip_address, attempt_number, validator_id, source, check_type, target, success, latency_ms, detail, checked_at, created_at FROM checks ` // ListChecksForAttempt returns every check recorded for an IP's current // attempt — the input to overall-result aggregation. func (d *DB) ListChecksForAttempt(ctx context.Context, ipID int64, attemptNumber int) ([]Check, error) { rows, err := d.QueryContext(ctx, checksSelect+` WHERE ip_id=? AND attempt_number=? ORDER BY source, check_type, target `, ipID, attemptNumber) if err != nil { return nil, err } defer rows.Close() return scanChecks(rows) } // ListChecksForRegistry returns an address's check history across every // cycle still retained (see PruneRegistryHistory), newest cycle first. A // nil limit returns everything currently retained. func (d *DB) ListChecksForRegistry(ctx context.Context, registryID int64, limit *int) ([]Check, error) { query := checksSelect + `WHERE registry_id=? ORDER BY cycle_id DESC, source, check_type, target` args := []interface{}{registryID} if limit != nil { query += ` LIMIT ?` args = append(args, *limit) } rows, err := d.QueryContext(ctx, query, args...) if err != nil { return nil, err } defer rows.Close() return scanChecks(rows) } func scanChecks(rows *sql.Rows) ([]Check, error) { var out []Check for rows.Next() { var c Check var ipID sql.NullInt64 var checkedAt, createdAt string if err := rows.Scan(&c.ID, &c.RegistryID, &c.CycleID, &ipID, &c.IPAddress, &c.AttemptNumber, &c.ValidatorID, &c.Source, &c.CheckType, &c.Target, &c.Success, &c.LatencyMS, &c.Detail, &checkedAt, &createdAt); err != nil { return nil, err } if ipID.Valid { c.IPID = ipID.Int64 } var err error if c.CheckedAt, err = dbToTime(checkedAt); err != nil { return nil, err } if c.CreatedAt, err = dbToTime(createdAt); err != nil { return nil, err } out = append(out, c) } return out, rows.Err() }