archivmail/internal/reconciliation/query.go

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()
}