story/internal/billing/overview.go

package billing

import (
	"context"
	"errors"
	"sort"
	"time"

	"data-terminal-cloud/internal/identity"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
)

const maxUsageOverviewRange = 366 * 24 * time.Hour

type UsageOverviewScope struct {
	TenantID      string `json:"tenant_id"`
	TenantName    string `json:"tenant_name"`
	HouseholdID   string `json:"household_id,omitempty"`
	HouseholdName string `json:"household_name,omitempty"`
	DeviceID      string `json:"device_id,omitempty"`
	DeviceName    string `json:"device_name,omitempty"`
}

type UsageOverviewDay struct {
	Date              string `json:"date"`
	DurationMS        int64  `json:"duration_ms"`
	ConversationCount int    `json:"conversation_count"`
	CostMicros        int64  `json:"cost_micros"`
}

type UsageOverviewDevice struct {
	ID                string `json:"id"`
	Name              string `json:"name"`
	HouseholdID       string `json:"household_id"`
	HouseholdName     string `json:"household_name"`
	DurationMS        int64  `json:"duration_ms"`
	ConversationCount int    `json:"conversation_count"`
	CostMicros        int64  `json:"cost_micros"`
}

type UsageOverviewConversation struct {
	ID            string     `json:"id"`
	HouseholdID   string     `json:"household_id"`
	HouseholdName string     `json:"household_name"`
	DeviceID      string     `json:"device_id"`
	DeviceName    string     `json:"device_name"`
	SubjectName   string     `json:"subject_name"`
	Status        string     `json:"status"`
	StartedAt     time.Time  `json:"started_at"`
	EndedAt       *time.Time `json:"ended_at,omitempty"`
	DurationMS    int64      `json:"duration_ms"`
	CostMicros    int64      `json:"cost_micros"`
}

type UsageOverview struct {
	From                time.Time                   `json:"from"`
	To                  time.Time                   `json:"to"`
	Timezone            string                      `json:"timezone"`
	Scope               UsageOverviewScope          `json:"scope"`
	TotalDurationMS     int64                       `json:"total_duration_ms"`
	TodayDurationMS     int64                       `json:"today_duration_ms"`
	ConversationCount   int                         `json:"conversation_count"`
	AverageDurationMS   int64                       `json:"average_duration_ms"`
	ActiveDays          int                         `json:"active_days"`
	CurrentStreakDays   int                         `json:"current_streak_days"`
	LongestStreakDays   int                         `json:"longest_streak_days"`
	CostMicros          int64                       `json:"cost_micros"`
	Currency            string                      `json:"currency"`
	CostAvailable       bool                        `json:"cost_available"`
	Daily               []UsageOverviewDay          `json:"daily"`
	Devices             []UsageOverviewDevice       `json:"devices"`
	RecentConversations []UsageOverviewConversation `json:"recent_conversations"`
}

type usageOverviewTarget struct {
	tenantID, tenantPublicID, tenantName          string
	householdID, householdPublicID, householdName string
	deviceID, devicePublicID, deviceName          string
}

type usageOverviewRow struct {
	conversationID, householdID, householdName string
	deviceID, deviceName, subjectName, status  string
	startedAt                                  time.Time
	endedAt                                    *time.Time
	costMicros                                 int64
	currency                                   string
	hasCost                                    bool
	isCallSession                              bool
}

func (s *Service) FamilyUsageOverview(ctx context.Context, principal identity.Principal, householdPublicID, devicePublicID, timezone string, from, to time.Time) (UsageOverview, error) {
	if _, err := uuid.Parse(householdPublicID); err != nil {
		return UsageOverview{}, ErrInvalid
	}
	target := usageOverviewTarget{}
	var role identity.Role
	err := s.pool.QueryRow(ctx, `
		SELECT t.id::text,t.public_id::text,t.name,h.id::text,h.public_id::text,h.name,hm.role
		FROM user_account ua
		JOIN tenant t ON t.id=ua.tenant_id AND t.status='active'
		JOIN household_member hm ON hm.user_id=ua.id AND hm.tenant_id=ua.tenant_id AND hm.status='active'
		JOIN household h ON h.id=hm.household_id AND h.tenant_id=hm.tenant_id AND h.status<>'deleted'
		WHERE h.public_id=$1::uuid AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3 AND ua.status='active'`,
		householdPublicID, principal.Issuer, principal.Subject).Scan(
		&target.tenantID, &target.tenantPublicID, &target.tenantName,
		&target.householdID, &target.householdPublicID, &target.householdName, &role)
	if errors.Is(err, pgx.ErrNoRows) || (err == nil && !identity.Can(role, identity.CapabilityHouseholdRead)) {
		return UsageOverview{}, ErrAccessDenied
	}
	if err != nil {
		return UsageOverview{}, err
	}
	if err = s.resolveUsageOverviewDevice(ctx, &target, devicePublicID, true); err != nil {
		return UsageOverview{}, err
	}
	return s.buildUsageOverview(ctx, target, timezone, from, to)
}

func (s *Service) OpsUsageOverview(ctx context.Context, principal identity.Principal, tenantPublicID, householdPublicID, devicePublicID, timezone string, from, to time.Time) (UsageOverview, error) {
	currentActor, err := s.resolveActor(ctx, principal, tenantPublicID, "ops_viewer", "ops_operator", "ops_approver", "ops_admin")
	if err != nil {
		return UsageOverview{}, err
	}
	target := usageOverviewTarget{tenantID: currentActor.tenantID, tenantPublicID: currentActor.tenantPublicID}
	if err = s.pool.QueryRow(ctx, `SELECT name FROM tenant WHERE id=$1::uuid`, target.tenantID).Scan(&target.tenantName); err != nil {
		return UsageOverview{}, err
	}
	if householdPublicID != "" {
		if _, parseErr := uuid.Parse(householdPublicID); parseErr != nil {
			return UsageOverview{}, ErrInvalid
		}
		err = s.pool.QueryRow(ctx, `SELECT id::text,public_id::text,name FROM household WHERE tenant_id=$1::uuid AND public_id=$2::uuid AND status<>'deleted'`, target.tenantID, householdPublicID).Scan(&target.householdID, &target.householdPublicID, &target.householdName)
		if errors.Is(err, pgx.ErrNoRows) {
			return UsageOverview{}, ErrNotFound
		}
		if err != nil {
			return UsageOverview{}, err
		}
	}
	if err = s.resolveUsageOverviewDevice(ctx, &target, devicePublicID, false); err != nil {
		return UsageOverview{}, err
	}
	return s.buildUsageOverview(ctx, target, timezone, from, to)
}

func (s *Service) resolveUsageOverviewDevice(ctx context.Context, target *usageOverviewTarget, devicePublicID string, family bool) error {
	if devicePublicID == "" {
		return nil
	}
	if _, err := uuid.Parse(devicePublicID); err != nil {
		return ErrInvalid
	}
	query := `SELECT d.id::text,d.public_id::text,d.model FROM device d WHERE d.tenant_id=$1::uuid AND d.public_id=$2::uuid`
	args := []any{target.tenantID, devicePublicID}
	if target.householdID != "" {
		query += ` AND (
			EXISTS(SELECT 1 FROM device_binding b WHERE b.tenant_id=d.tenant_id AND b.device_id=d.id AND b.household_id=$3::uuid AND b.unbound_at IS NULL)
			OR EXISTS(SELECT 1 FROM conversation c WHERE c.tenant_id=d.tenant_id AND c.device_id=d.id AND c.household_id=$3::uuid)
			OR EXISTS(SELECT 1 FROM call_session c JOIN subject_profile sp ON sp.id=c.subject_id AND sp.tenant_id=c.tenant_id
				WHERE c.tenant_id=d.tenant_id AND c.device_id=d.id AND sp.household_id=$3::uuid)
		)`
		args = append(args, target.householdID)
	}
	err := s.pool.QueryRow(ctx, query, args...).Scan(&target.deviceID, &target.devicePublicID, &target.deviceName)
	if errors.Is(err, pgx.ErrNoRows) {
		if family {
			return ErrAccessDenied
		}
		return ErrNotFound
	}
	return err
}

func (s *Service) buildUsageOverview(ctx context.Context, target usageOverviewTarget, timezone string, from, to time.Time) (UsageOverview, error) {
	location, err := validateUsageOverviewRange(timezone, from, to)
	if err != nil {
		return UsageOverview{}, err
	}
	from = from.UTC()
	to = to.UTC()
	rows, err := s.pool.Query(ctx, `
		SELECT c.public_id::text AS conversation_id,h.public_id::text,h.name,d.public_id::text,d.model,sp.preferred_name,c.status,c.started_at,c.ended_at,
			COALESCE(cost.cost_micros,0)::bigint,COALESCE(cost.currency,''),COALESCE(cost.has_cost,false),false AS is_call_session
		FROM conversation c
		JOIN household h ON h.id=c.household_id AND h.tenant_id=c.tenant_id
		JOIN device d ON d.id=c.device_id AND d.tenant_id=c.tenant_id
		JOIN subject_profile sp ON sp.id=c.subject_id AND sp.tenant_id=c.tenant_id
		LEFT JOIN LATERAL (
			SELECT SUM(ccl.cost_micros)::bigint AS cost_micros,MIN(ccl.currency) AS currency,COUNT(*)>0 AS has_cost
			FROM conversation_cost_line ccl WHERE ccl.conversation_id=c.id
		) cost ON true
		WHERE c.tenant_id=$1::uuid AND c.started_at IS NOT NULL AND c.started_at>=$2 AND c.started_at<$3
			AND ($4::uuid IS NULL OR c.household_id=$4::uuid)
			AND ($5::uuid IS NULL OR c.device_id=$5::uuid)
		UNION ALL
		SELECT c.id::text,h.public_id::text,h.name,d.public_id::text,d.model,sp.preferred_name,
			CASE c.call_state WHEN 'ENDED' THEN 'COMPLETED' WHEN 'TALKING' THEN 'ACTIVE'
				WHEN 'CANCELLED' THEN 'FAILED' WHEN 'STARTING' THEN 'STARTING_AGENT' ELSE c.call_state END,
			c.created_at,c.ended_at,0::bigint,''::text,false,true
		FROM call_session c
		JOIN subject_profile sp ON sp.id=c.subject_id AND sp.tenant_id=c.tenant_id
		JOIN household h ON h.id=sp.household_id AND h.tenant_id=c.tenant_id
		JOIN device d ON d.id=c.device_id AND d.tenant_id=c.tenant_id
		WHERE c.tenant_id=$1::uuid AND c.created_at>=$2 AND c.created_at<$3
			AND ($4::uuid IS NULL OR sp.household_id=$4::uuid)
			AND ($5::uuid IS NULL OR c.device_id=$5::uuid)
		ORDER BY started_at DESC,conversation_id`, target.tenantID, from, to, nullableUUID(target.householdID), nullableUUID(target.deviceID))
	if err != nil {
		return UsageOverview{}, err
	}
	defer rows.Close()

	result := UsageOverview{
		From: from, To: to, Timezone: timezone,
		Scope: UsageOverviewScope{TenantID: target.tenantPublicID, TenantName: target.tenantName, HouseholdID: target.householdPublicID, HouseholdName: target.householdName, DeviceID: target.devicePublicID, DeviceName: target.deviceName},
		Daily: []UsageOverviewDay{}, Devices: []UsageOverviewDevice{}, RecentConversations: []UsageOverviewConversation{},
	}
	daily := map[string]*UsageOverviewDay{}
	devices := map[string]*UsageOverviewDevice{}
	activeDates := map[string]struct{}{}
	currencies := map[string]struct{}{}
	today := time.Now().In(location).Format("2006-01-02")
	for rows.Next() {
		var row usageOverviewRow
		if err = rows.Scan(&row.conversationID, &row.householdID, &row.householdName, &row.deviceID, &row.deviceName, &row.subjectName, &row.status, &row.startedAt, &row.endedAt, &row.costMicros, &row.currency, &row.hasCost, &row.isCallSession); err != nil {
			return UsageOverview{}, err
		}
		duration := conversationDurationMS(row.startedAt, row.endedAt, to)
		// Match calls/query: a new session has no finalized duration until it ends.
		// Do not inflate abandoned preparations or missing end reports up to now.
		if row.isCallSession && row.endedAt == nil {
			duration = 0
		}
		date := row.startedAt.In(location).Format("2006-01-02")
		day := daily[date]
		if day == nil {
			day = &UsageOverviewDay{Date: date}
			daily[date] = day
		}
		day.DurationMS += duration
		day.ConversationCount++
		day.CostMicros += row.costMicros
		activeDates[date] = struct{}{}

		device := devices[row.deviceID]
		if device == nil {
			device = &UsageOverviewDevice{ID: row.deviceID, Name: row.deviceName, HouseholdID: row.householdID, HouseholdName: row.householdName}
			devices[row.deviceID] = device
		}
		device.DurationMS += duration
		device.ConversationCount++
		device.CostMicros += row.costMicros

		result.TotalDurationMS += duration
		result.ConversationCount++
		result.CostMicros += row.costMicros
		if date == today {
			result.TodayDurationMS += duration
		}
		if row.hasCost {
			result.CostAvailable = true
			currencies[row.currency] = struct{}{}
		}
		if len(result.RecentConversations) < 20 {
			result.RecentConversations = append(result.RecentConversations, UsageOverviewConversation{
				ID: row.conversationID, HouseholdID: row.householdID, HouseholdName: row.householdName,
				DeviceID: row.deviceID, DeviceName: row.deviceName, SubjectName: row.subjectName, Status: row.status,
				StartedAt: row.startedAt, EndedAt: row.endedAt, DurationMS: duration, CostMicros: row.costMicros,
			})
		}
	}
	if err = rows.Err(); err != nil {
		return UsageOverview{}, err
	}
	if len(currencies) > 1 {
		return UsageOverview{}, errors.New("usage overview contains multiple currencies")
	}
	for currency := range currencies {
		result.Currency = currency
	}
	if result.ConversationCount > 0 {
		result.AverageDurationMS = result.TotalDurationMS / int64(result.ConversationCount)
	}
	for _, day := range daily {
		result.Daily = append(result.Daily, *day)
	}
	sort.Slice(result.Daily, func(i, j int) bool { return result.Daily[i].Date < result.Daily[j].Date })
	for _, device := range devices {
		result.Devices = append(result.Devices, *device)
	}
	sort.Slice(result.Devices, func(i, j int) bool {
		if result.Devices[i].DurationMS == result.Devices[j].DurationMS {
			return result.Devices[i].ID < result.Devices[j].ID
		}
		return result.Devices[i].DurationMS > result.Devices[j].DurationMS
	})
	result.ActiveDays = len(activeDates)
	result.CurrentStreakDays, result.LongestStreakDays = usageStreaks(activeDates, time.Now().In(location))
	return result, nil
}

func validateUsageOverviewRange(timezone string, from, to time.Time) (*time.Location, error) {
	if timezone == "" || from.IsZero() || !to.After(from) || to.Sub(from) > maxUsageOverviewRange || to.After(time.Now().UTC().Add(5*time.Minute)) {
		return nil, ErrInvalid
	}
	location, err := time.LoadLocation(timezone)
	if err != nil {
		return nil, ErrInvalid
	}
	return location, nil
}

func conversationDurationMS(startedAt time.Time, endedAt *time.Time, rangeEnd time.Time) int64 {
	end := rangeEnd
	if endedAt != nil && endedAt.Before(end) {
		end = *endedAt
	}
	if end.Before(startedAt) {
		return 0
	}
	return end.Sub(startedAt).Milliseconds()
}

func usageStreaks(active map[string]struct{}, now time.Time) (current, longest int) {
	if len(active) == 0 {
		return 0, 0
	}
	dates := make([]time.Time, 0, len(active))
	for value := range active {
		date, err := time.ParseInLocation("2006-01-02", value, now.Location())
		if err == nil {
			dates = append(dates, date)
		}
	}
	sort.Slice(dates, func(i, j int) bool { return dates[i].Before(dates[j]) })
	streak := 0
	for index, date := range dates {
		if index == 0 || date.Equal(dates[index-1].AddDate(0, 0, 1)) {
			streak++
		} else {
			streak = 1
		}
		if streak > longest {
			longest = streak
		}
	}
	latest := dates[len(dates)-1]
	today, _ := time.ParseInLocation("2006-01-02", now.Format("2006-01-02"), now.Location())
	if latest.Before(today.AddDate(0, 0, -1)) {
		return 0, longest
	}
	current = 1
	for index := len(dates) - 1; index > 0 && dates[index].Equal(dates[index-1].AddDate(0, 0, 1)); index-- {
		current++
	}
	return current, longest
}

func nullableUUID(value string) any {
	if value == "" {
		return nil
	}
	return value
}