package reconciliation import ( "context" "fmt" "time" ) // DayPoint is one day's figures for a source in the dashboard response. // ArchivedCount is nil (and Missing true) when no report row exists for that // date — i.e. the cron job did not run — which is distinct from an archived // count of 0 on a genuine zero-activity day. type DayPoint struct { Date string `json:"date"` ArchivedCount *int64 `json:"archived_count"` ExpectedCount *int64 `json:"expected_count"` Delta *int64 `json:"delta"` Missing bool `json:"missing"` } // SourceSummary aggregates the trailing days plus alert state for one source. type SourceSummary struct { SourceType string `json:"source_type"` SourceID *int64 `json:"source_id"` SourceKey string `json:"source_key"` TenantID *int64 `json:"tenant_id"` Points []DayPoint `json:"points"` Avg7d float64 `json:"avg_7d"` EnoughData bool `json:"enough_data"` Alert bool `json:"alert"` } // DashboardData returns the last `days` calendar days of reconciliation figures // per source, tenant-scoped. When tenantID is nil (superadmin) all tenants are // included; otherwise only rows for that tenant are returned. thresholdPct is // used to compute the per-source Alert flag consistently with the cron job. func (s *Store) DashboardData(ctx context.Context, tenantID *int64, days, thresholdPct int) ([]SourceSummary, error) { if days <= 0 { days = 7 } today := time.Now().UTC().Truncate(24 * time.Hour) start := today.AddDate(0, 0, -(days - 1)) rows, err := s.queryRange(ctx, tenantID, start, today.Add(24*time.Hour)) if err != nil { return nil, err } // Ordered date labels for the window. dateLabels := make([]string, days) for i := 0; i < days; i++ { dateLabels[i] = start.AddDate(0, 0, i).Format("2006-01-02") } type srcAgg struct { meta Report byDate map[string]Report } agg := map[string]*srcAgg{} for _, r := range rows { key := SourceKey(r.SourceType, r.SourceID) // Distinguish sources of different tenants sharing a key. if r.TenantID != nil { key = fmt.Sprintf("t%d/%s", *r.TenantID, key) } a, ok := agg[key] if !ok { a = &srcAgg{meta: r, byDate: map[string]Report{}} agg[key] = a } a.byDate[r.Date.UTC().Format("2006-01-02")] = r } summaries := make([]SourceSummary, 0, len(agg)) for _, a := range agg { sum := SourceSummary{ SourceType: a.meta.SourceType, SourceID: a.meta.SourceID, SourceKey: SourceKey(a.meta.SourceType, a.meta.SourceID), TenantID: a.meta.TenantID, Points: make([]DayPoint, 0, days), } for _, d := range dateLabels { if r, ok := a.byDate[d]; ok { c := r.ArchivedCount sum.Points = append(sum.Points, DayPoint{ Date: d, ArchivedCount: &c, ExpectedCount: r.ExpectedCount, Delta: r.Delta, Missing: false, }) } else { sum.Points = append(sum.Points, DayPoint{Date: d, Missing: true}) } } // Alert against the trailing 7-day average of the most recent day that // actually has a report row (mirrors the cron job's evaluation). latest, latestDate, hasLatest := latestPresent(a.byDate, dateLabels) if hasLatest { avg, n, err := s.trailingAverage(ctx, a.meta, latestDate) if err == nil && n >= 7 { sum.Avg7d = avg sum.EnoughData = true limit := avg * (1 - float64(thresholdPct)/100.0) if float64(latest.ArchivedCount) < limit { sum.Alert = true } } } summaries = append(summaries, sum) } return summaries, nil } // latestPresent returns the most recent report row within the window that has a // stored row, along with its date. func latestPresent(byDate map[string]Report, dateLabels []string) (Report, time.Time, bool) { for i := len(dateLabels) - 1; i >= 0; i-- { if r, ok := byDate[dateLabels[i]]; ok { d, _ := time.Parse("2006-01-02", dateLabels[i]) return r, d, true } } return Report{}, time.Time{}, false } // ExportRows returns raw report rows for CSV export, tenant-scoped, for the // last `days` days, ordered by date descending then source. func (s *Store) ExportRows(ctx context.Context, tenantID *int64, days int) ([]Report, error) { if days <= 0 { days = 30 } today := time.Now().UTC().Truncate(24 * time.Hour) start := today.AddDate(0, 0, -(days - 1)) return s.queryRange(ctx, tenantID, start, today.Add(24*time.Hour)) } // queryRange loads report rows in [start, end) filtered by tenant. tenantID nil // returns all tenants (superadmin scope). func (s *Store) queryRange(ctx context.Context, tenantID *int64, start, end time.Time) ([]Report, error) { var ( sql string args []interface{} ) if tenantID == nil { sql = `SELECT date, tenant_id, source_type, source_id, expected_count, archived_count, delta FROM reconciliation_reports WHERE date >= $1 AND date < $2 ORDER BY date DESC, source_type, source_id` args = []interface{}{start, end} } else { sql = `SELECT date, tenant_id, source_type, source_id, expected_count, archived_count, delta FROM reconciliation_reports WHERE date >= $1 AND date < $2 AND tenant_id = $3 ORDER BY date DESC, source_type, source_id` args = []interface{}{start, end, *tenantID} } rows, err := s.pool.Query(ctx, sql, args...) if err != nil { return nil, fmt.Errorf("reconciliation: query range: %w", err) } defer rows.Close() var out []Report for rows.Next() { var r Report if err := rows.Scan(&r.Date, &r.TenantID, &r.SourceType, &r.SourceID, &r.ExpectedCount, &r.ArchivedCount, &r.Delta); err != nil { return nil, fmt.Errorf("reconciliation: scan range: %w", err) } out = append(out, r) } return out, rows.Err() }