497 lines
18 KiB
Go
497 lines
18 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
type monitoringBrandDailyAggregate struct {
|
|
PlannedSampleCount int64
|
|
ActualSampleCount int64
|
|
MentionedCount int64
|
|
Top1MentionedCount int64
|
|
FirstRecommendedCount int64
|
|
PositiveMentionedCount int64
|
|
CitedAnswerCount int64
|
|
SnapshotUpdatedAt sql.NullTime
|
|
LatestTriggerSource sql.NullString
|
|
}
|
|
|
|
type monitoringPlatformDailyAggregate struct {
|
|
AIPlatformID string
|
|
PlannedSampleCount int64
|
|
ActualSampleCount int64
|
|
MentionedCount int64
|
|
Top1MentionedCount int64
|
|
LastSampledAt sql.NullTime
|
|
AccessStatus string
|
|
}
|
|
|
|
func (s *MonitoringCallbackService) rebuildBrandDailyProjection(ctx context.Context, tx pgx.Tx, tenantID, brandID int64, businessDate time.Time) error {
|
|
platforms, err := loadMonitoringProjectionPlatforms(ctx, tx, tenantID, brandID, businessDate)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
enabledQuestionCount, err := s.loadMonitoringEnabledQuestionCount(ctx, tx, tenantID, brandID, businessDate)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
brandAgg, err := s.loadMonitoringBrandDailyAggregate(ctx, tx, tenantID, brandID, businessDate)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
platformAggs, err := s.loadMonitoringPlatformDailyAggregates(ctx, tx, tenantID, brandID, businessDate, platforms)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
desiredSampleCount := enabledQuestionCount * int64(len(platforms))
|
|
coverageRate := divideAsPointer(brandAgg.ActualSampleCount, brandAgg.PlannedSampleCount)
|
|
mentionRate := divideAsPointer(brandAgg.MentionedCount, brandAgg.ActualSampleCount)
|
|
top1MentionRate := divideAsPointer(brandAgg.Top1MentionedCount, brandAgg.ActualSampleCount)
|
|
firstRecommendRate := divideAsPointer(brandAgg.FirstRecommendedCount, brandAgg.ActualSampleCount)
|
|
positiveMentionRate := divideAsPointer(brandAgg.PositiveMentionedCount, brandAgg.ActualSampleCount)
|
|
citationRate := divideAsPointer(brandAgg.CitedAnswerCount, brandAgg.ActualSampleCount)
|
|
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO monitoring_brand_daily (
|
|
tenant_id, brand_id, collector_type, business_date, sample_status,
|
|
desired_sample_count, planned_sample_count, actual_sample_count,
|
|
coverage_rate, confidence_level, trigger_source, snapshot_updated_at,
|
|
mentioned_count, top1_mentioned_count, first_recommended_count, positive_mentioned_count,
|
|
cited_answer_count, mention_rate, top1_mention_rate, first_recommend_rate,
|
|
positive_mention_rate, citation_rate
|
|
)
|
|
VALUES (
|
|
$1, $2, $3, $4::date, $5,
|
|
$6, $7, $8,
|
|
$9, $10, $11, $12,
|
|
$13, $14, $15, $16,
|
|
$17, $18, $19, $20,
|
|
$21, $22
|
|
)
|
|
ON CONFLICT (tenant_id, brand_id, collector_type, business_date)
|
|
DO UPDATE SET
|
|
sample_status = EXCLUDED.sample_status,
|
|
desired_sample_count = EXCLUDED.desired_sample_count,
|
|
planned_sample_count = EXCLUDED.planned_sample_count,
|
|
actual_sample_count = EXCLUDED.actual_sample_count,
|
|
coverage_rate = EXCLUDED.coverage_rate,
|
|
confidence_level = EXCLUDED.confidence_level,
|
|
trigger_source = EXCLUDED.trigger_source,
|
|
snapshot_updated_at = EXCLUDED.snapshot_updated_at,
|
|
mentioned_count = EXCLUDED.mentioned_count,
|
|
top1_mentioned_count = EXCLUDED.top1_mentioned_count,
|
|
first_recommended_count = EXCLUDED.first_recommended_count,
|
|
positive_mentioned_count = EXCLUDED.positive_mentioned_count,
|
|
cited_answer_count = EXCLUDED.cited_answer_count,
|
|
mention_rate = EXCLUDED.mention_rate,
|
|
top1_mention_rate = EXCLUDED.top1_mention_rate,
|
|
first_recommend_rate = EXCLUDED.first_recommend_rate,
|
|
positive_mention_rate = EXCLUDED.positive_mention_rate,
|
|
citation_rate = EXCLUDED.citation_rate,
|
|
updated_at = NOW()
|
|
`,
|
|
tenantID,
|
|
brandID,
|
|
monitoringCollectorType,
|
|
monitoringBusinessDateText(businessDate),
|
|
deriveMonitoringBrandSampleStatus(brandAgg.ActualSampleCount, brandAgg.PlannedSampleCount, coverageRate),
|
|
desiredSampleCount,
|
|
brandAgg.PlannedSampleCount,
|
|
brandAgg.ActualSampleCount,
|
|
nullableFloat64(coverageRate),
|
|
deriveMonitoringConfidenceLevel(brandAgg.ActualSampleCount, brandAgg.PlannedSampleCount, coverageRate),
|
|
nullableString(stringPointer(brandAgg.LatestTriggerSource)),
|
|
nullableTime(brandAgg.SnapshotUpdatedAt),
|
|
brandAgg.MentionedCount,
|
|
brandAgg.Top1MentionedCount,
|
|
brandAgg.FirstRecommendedCount,
|
|
brandAgg.PositiveMentionedCount,
|
|
brandAgg.CitedAnswerCount,
|
|
nullableFloat64(mentionRate),
|
|
nullableFloat64(top1MentionRate),
|
|
nullableFloat64(firstRecommendRate),
|
|
nullableFloat64(positiveMentionRate),
|
|
nullableFloat64(citationRate),
|
|
); err != nil {
|
|
return monitoringInternalError(50041, "brand_daily_projection_failed", "failed to persist monitoring brand daily projection", err)
|
|
}
|
|
|
|
if err := s.replaceMonitoringPlatformDailyProjection(ctx, tx, tenantID, brandID, businessDate, platformAggs); err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func loadMonitoringProjectionPlatforms(_ context.Context, _ pgx.Tx, _ int64, _ int64, _ time.Time) ([]monitoringPlatformMetadata, error) {
|
|
return defaultMonitoringPlatformMetadata(), nil
|
|
}
|
|
|
|
func (s *MonitoringCallbackService) loadMonitoringEnabledQuestionCount(ctx context.Context, tx pgx.Tx, tenantID, brandID int64, businessDate time.Time) (int64, error) {
|
|
dayEnd := monitoringCanonicalBusinessDay(businessDate).AddDate(0, 0, 1)
|
|
|
|
var count int64
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT COUNT(DISTINCT question_id)
|
|
FROM monitoring_question_config_snapshots
|
|
WHERE tenant_id = $1
|
|
AND brand_id = $2
|
|
AND monitor_enabled = TRUE
|
|
AND projected_at < $3
|
|
AND (superseded_at IS NULL OR superseded_at >= $3)
|
|
AND (deleted_at IS NULL OR deleted_at >= $3)
|
|
`, tenantID, brandID, dayEnd).Scan(&count); err != nil {
|
|
return 0, monitoringInternalError(50041, "question_snapshot_lookup_failed", "failed to load monitoring question snapshots", err)
|
|
}
|
|
|
|
return count, nil
|
|
}
|
|
|
|
func (s *MonitoringCallbackService) loadMonitoringBrandDailyAggregate(ctx context.Context, tx pgx.Tx, tenantID, brandID int64, businessDate time.Time) (monitoringBrandDailyAggregate, error) {
|
|
agg := monitoringBrandDailyAggregate{}
|
|
dateText := monitoringBusinessDateText(businessDate)
|
|
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT COUNT(*)
|
|
FROM monitoring_collect_tasks
|
|
WHERE tenant_id = $1
|
|
AND brand_id = $2
|
|
AND collector_type = $3
|
|
AND business_date = $4::date
|
|
AND NOT (status = 'skipped' AND skip_reason = ANY($5::text[]))
|
|
`, tenantID, brandID, monitoringCollectorType, dateText, monitoringProjectionExcludedSkipReasons).Scan(&agg.PlannedSampleCount); err != nil {
|
|
return agg, monitoringInternalError(50041, "task_projection_lookup_failed", "failed to load monitoring task counts", err)
|
|
}
|
|
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT COUNT(*), MAX(completed_at)
|
|
FROM question_monitor_runs
|
|
WHERE tenant_id = $1
|
|
AND brand_id = $2
|
|
AND collector_type = $3
|
|
AND business_date = $4::date
|
|
AND status = 'succeeded'
|
|
`, tenantID, brandID, monitoringCollectorType, dateText).Scan(&agg.ActualSampleCount, &agg.SnapshotUpdatedAt); err != nil {
|
|
return agg, monitoringInternalError(50041, "run_projection_lookup_failed", "failed to load monitoring run counts", err)
|
|
}
|
|
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT
|
|
COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE)::bigint,
|
|
COUNT(*) FILTER (WHERE p.brand_mention_position = 'top1')::bigint,
|
|
COUNT(*) FILTER (WHERE p.first_recommended IS TRUE)::bigint,
|
|
COUNT(*) FILTER (WHERE p.sentiment_label = 'positive')::bigint
|
|
FROM question_monitor_runs r
|
|
LEFT JOIN question_monitor_parse_results p
|
|
ON p.tenant_id = r.tenant_id AND p.run_id = r.id
|
|
WHERE r.tenant_id = $1
|
|
AND r.brand_id = $2
|
|
AND r.collector_type = $3
|
|
AND r.business_date = $4::date
|
|
AND r.status = 'succeeded'
|
|
`, tenantID, brandID, monitoringCollectorType, dateText).Scan(
|
|
&agg.MentionedCount,
|
|
&agg.Top1MentionedCount,
|
|
&agg.FirstRecommendedCount,
|
|
&agg.PositiveMentionedCount,
|
|
); err != nil {
|
|
return agg, monitoringInternalError(50041, "parse_projection_lookup_failed", "failed to load monitoring parse aggregates", err)
|
|
}
|
|
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT COUNT(DISTINCT cf.run_id)
|
|
FROM monitoring_citation_facts cf
|
|
JOIN question_monitor_runs r
|
|
ON r.tenant_id = cf.tenant_id AND r.id = cf.run_id
|
|
WHERE cf.tenant_id = $1
|
|
AND cf.brand_id = $2
|
|
AND cf.business_date = $4::date
|
|
AND cf.content_source_type IS NOT NULL
|
|
AND cf.content_source_id IS NOT NULL
|
|
AND r.collector_type = $3
|
|
AND r.business_date = $4::date
|
|
AND r.status = 'succeeded'
|
|
`, tenantID, brandID, monitoringCollectorType, dateText).Scan(&agg.CitedAnswerCount); err != nil {
|
|
return agg, monitoringInternalError(50041, "citation_projection_lookup_failed", "failed to load monitoring citation aggregates", err)
|
|
}
|
|
|
|
if err := tx.QueryRow(ctx, `
|
|
SELECT trigger_source
|
|
FROM question_monitor_runs
|
|
WHERE tenant_id = $1
|
|
AND brand_id = $2
|
|
AND collector_type = $3
|
|
AND business_date = $4::date
|
|
AND status = 'succeeded'
|
|
ORDER BY completed_at DESC NULLS LAST, id DESC
|
|
LIMIT 1
|
|
`, tenantID, brandID, monitoringCollectorType, dateText).Scan(&agg.LatestTriggerSource); err != nil && err != pgx.ErrNoRows {
|
|
return agg, monitoringInternalError(50041, "trigger_source_lookup_failed", "failed to load monitoring trigger source", err)
|
|
}
|
|
|
|
return agg, nil
|
|
}
|
|
|
|
func (s *MonitoringCallbackService) loadMonitoringPlatformDailyAggregates(ctx context.Context, tx pgx.Tx, tenantID, brandID int64, businessDate time.Time, platforms []monitoringPlatformMetadata) ([]monitoringPlatformDailyAggregate, error) {
|
|
dateText := monitoringBusinessDateText(businessDate)
|
|
|
|
taskCounts := make(map[string]int64, len(platforms))
|
|
taskRows, err := tx.Query(ctx, `
|
|
SELECT ai_platform_id, COUNT(*)
|
|
FROM monitoring_collect_tasks
|
|
WHERE tenant_id = $1
|
|
AND brand_id = $2
|
|
AND collector_type = $3
|
|
AND business_date = $4::date
|
|
AND NOT (status = 'skipped' AND skip_reason = ANY($5::text[]))
|
|
GROUP BY ai_platform_id
|
|
`, tenantID, brandID, monitoringCollectorType, dateText, monitoringProjectionExcludedSkipReasons)
|
|
if err != nil {
|
|
return nil, monitoringInternalError(50041, "platform_task_projection_lookup_failed", "failed to load monitoring platform task counts", err)
|
|
}
|
|
for taskRows.Next() {
|
|
var platformID string
|
|
var plannedSampleCount int64
|
|
if scanErr := taskRows.Scan(&platformID, &plannedSampleCount); scanErr != nil {
|
|
taskRows.Close()
|
|
return nil, monitoringInternalError(50041, "platform_task_projection_scan_failed", "failed to parse monitoring platform task counts", scanErr)
|
|
}
|
|
taskCounts[platformID] = plannedSampleCount
|
|
}
|
|
if err := taskRows.Err(); err != nil {
|
|
taskRows.Close()
|
|
return nil, monitoringInternalError(50041, "platform_task_projection_scan_failed", "failed to iterate monitoring platform task counts", err)
|
|
}
|
|
taskRows.Close()
|
|
|
|
runAggs := make(map[string]monitoringPlatformDailyAggregate, len(platforms))
|
|
runRows, err := tx.Query(ctx, `
|
|
SELECT
|
|
r.ai_platform_id,
|
|
COUNT(*)::bigint AS actual_sample_count,
|
|
COUNT(*) FILTER (WHERE p.brand_mentioned IS TRUE)::bigint AS mentioned_count,
|
|
COUNT(*) FILTER (WHERE p.brand_mention_position = 'top1')::bigint AS top1_mentioned_count,
|
|
MAX(r.completed_at) AS last_sampled_at
|
|
FROM question_monitor_runs r
|
|
LEFT JOIN question_monitor_parse_results p
|
|
ON p.tenant_id = r.tenant_id AND p.run_id = r.id
|
|
WHERE r.tenant_id = $1
|
|
AND r.brand_id = $2
|
|
AND r.collector_type = $3
|
|
AND r.business_date = $4::date
|
|
AND r.status = 'succeeded'
|
|
GROUP BY r.ai_platform_id
|
|
`, tenantID, brandID, monitoringCollectorType, dateText)
|
|
if err != nil {
|
|
return nil, monitoringInternalError(50041, "platform_run_projection_lookup_failed", "failed to load monitoring platform run aggregates", err)
|
|
}
|
|
for runRows.Next() {
|
|
var item monitoringPlatformDailyAggregate
|
|
if scanErr := runRows.Scan(
|
|
&item.AIPlatformID,
|
|
&item.ActualSampleCount,
|
|
&item.MentionedCount,
|
|
&item.Top1MentionedCount,
|
|
&item.LastSampledAt,
|
|
); scanErr != nil {
|
|
runRows.Close()
|
|
return nil, monitoringInternalError(50041, "platform_run_projection_scan_failed", "failed to parse monitoring platform run aggregates", scanErr)
|
|
}
|
|
runAggs[item.AIPlatformID] = item
|
|
}
|
|
if err := runRows.Err(); err != nil {
|
|
runRows.Close()
|
|
return nil, monitoringInternalError(50041, "platform_run_projection_scan_failed", "failed to iterate monitoring platform run aggregates", err)
|
|
}
|
|
runRows.Close()
|
|
|
|
accessStates := make(map[string]string, len(platforms))
|
|
accessRows, err := tx.Query(ctx, `
|
|
SELECT DISTINCT ON (ai_platform_id) ai_platform_id, access_status
|
|
FROM monitoring_platform_access_snapshots
|
|
WHERE tenant_id = $1
|
|
AND business_date = $2::date
|
|
ORDER BY ai_platform_id ASC, detected_at DESC, id DESC
|
|
`, tenantID, dateText)
|
|
if err != nil {
|
|
return nil, monitoringInternalError(50041, "platform_access_projection_lookup_failed", "failed to load monitoring platform access states", err)
|
|
}
|
|
for accessRows.Next() {
|
|
var platformID string
|
|
var accessStatus string
|
|
if scanErr := accessRows.Scan(&platformID, &accessStatus); scanErr != nil {
|
|
accessRows.Close()
|
|
return nil, monitoringInternalError(50041, "platform_access_projection_scan_failed", "failed to parse monitoring platform access states", scanErr)
|
|
}
|
|
accessStates[platformID] = accessStatus
|
|
}
|
|
if err := accessRows.Err(); err != nil {
|
|
accessRows.Close()
|
|
return nil, monitoringInternalError(50041, "platform_access_projection_scan_failed", "failed to iterate monitoring platform access states", err)
|
|
}
|
|
accessRows.Close()
|
|
|
|
items := make([]monitoringPlatformDailyAggregate, 0, len(platforms))
|
|
for _, platform := range platforms {
|
|
item := runAggs[platform.ID]
|
|
item.AIPlatformID = platform.ID
|
|
item.PlannedSampleCount = taskCounts[platform.ID]
|
|
item.AccessStatus = accessStates[platform.ID]
|
|
items = append(items, item)
|
|
}
|
|
|
|
return items, nil
|
|
}
|
|
|
|
func (s *MonitoringCallbackService) replaceMonitoringPlatformDailyProjection(ctx context.Context, tx pgx.Tx, tenantID, brandID int64, businessDate time.Time, items []monitoringPlatformDailyAggregate) error {
|
|
dateText := monitoringBusinessDateText(businessDate)
|
|
|
|
if _, err := tx.Exec(ctx, `
|
|
DELETE FROM monitoring_brand_platform_daily
|
|
WHERE tenant_id = $1
|
|
AND brand_id = $2
|
|
AND collector_type = $3
|
|
AND business_date = $4::date
|
|
`, tenantID, brandID, monitoringCollectorType, dateText); err != nil {
|
|
return monitoringInternalError(50041, "platform_daily_projection_cleanup_failed", "failed to reset monitoring platform daily projection", err)
|
|
}
|
|
|
|
for _, item := range items {
|
|
mentionRate := divideAsPointer(item.MentionedCount, item.ActualSampleCount)
|
|
top1MentionRate := divideAsPointer(item.Top1MentionedCount, item.ActualSampleCount)
|
|
|
|
if _, err := tx.Exec(ctx, `
|
|
INSERT INTO monitoring_brand_platform_daily (
|
|
tenant_id, brand_id, ai_platform_id, collector_type, business_date,
|
|
platform_sample_status, planned_sample_count, actual_sample_count,
|
|
mention_rate, top1_mention_rate, last_sampled_at
|
|
)
|
|
VALUES (
|
|
$1, $2, $3, $4, $5::date,
|
|
$6, $7, $8,
|
|
$9, $10, $11
|
|
)
|
|
`,
|
|
tenantID,
|
|
brandID,
|
|
item.AIPlatformID,
|
|
monitoringCollectorType,
|
|
dateText,
|
|
deriveMonitoringPlatformSampleStatus(item.ActualSampleCount, item.PlannedSampleCount, item.AccessStatus),
|
|
item.PlannedSampleCount,
|
|
item.ActualSampleCount,
|
|
nullableFloat64(mentionRate),
|
|
nullableFloat64(top1MentionRate),
|
|
nullableTime(item.LastSampledAt),
|
|
); err != nil {
|
|
return monitoringInternalError(50041, "platform_daily_projection_failed", "failed to persist monitoring platform daily projection", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *MonitoringCallbackService) loadMonitoringAffectedBrandIDs(ctx context.Context, tx pgx.Tx, tenantID int64, businessDate time.Time, aiPlatformID string) ([]int64, error) {
|
|
rows, err := tx.Query(ctx, `
|
|
SELECT DISTINCT brand_id
|
|
FROM monitoring_collect_tasks
|
|
WHERE tenant_id = $1
|
|
AND ai_platform_id = $2
|
|
AND collector_type = $3
|
|
AND business_date = $4::date
|
|
ORDER BY brand_id ASC
|
|
`, tenantID, aiPlatformID, monitoringCollectorType, monitoringBusinessDateText(businessDate))
|
|
if err != nil {
|
|
return nil, monitoringInternalError(50041, "affected_brand_lookup_failed", "failed to load monitoring brands for projection", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
brandIDs := make([]int64, 0)
|
|
for rows.Next() {
|
|
var brandID int64
|
|
if scanErr := rows.Scan(&brandID); scanErr != nil {
|
|
return nil, monitoringInternalError(50041, "affected_brand_scan_failed", "failed to parse monitoring brands for projection", scanErr)
|
|
}
|
|
brandIDs = append(brandIDs, brandID)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, monitoringInternalError(50041, "affected_brand_scan_failed", "failed to iterate monitoring brands for projection", err)
|
|
}
|
|
|
|
return brandIDs, nil
|
|
}
|
|
|
|
func deriveMonitoringBrandSampleStatus(actualSampleCount, plannedSampleCount int64, coverageRate *float64) string {
|
|
if actualSampleCount <= 0 {
|
|
return "unsampled"
|
|
}
|
|
if plannedSampleCount <= 0 {
|
|
return "sampled"
|
|
}
|
|
if coverageRate != nil && *coverageRate >= 0.7 {
|
|
return "sampled"
|
|
}
|
|
return "partial"
|
|
}
|
|
|
|
func deriveMonitoringConfidenceLevel(actualSampleCount, plannedSampleCount int64, coverageRate *float64) string {
|
|
if actualSampleCount <= 0 {
|
|
return "low"
|
|
}
|
|
if plannedSampleCount <= 0 {
|
|
return "high"
|
|
}
|
|
if coverageRate == nil {
|
|
return "low"
|
|
}
|
|
switch {
|
|
case *coverageRate >= 0.7:
|
|
return "high"
|
|
case *coverageRate >= 0.4:
|
|
return "medium"
|
|
default:
|
|
return "low"
|
|
}
|
|
}
|
|
|
|
func deriveMonitoringPlatformSampleStatus(actualSampleCount, plannedSampleCount int64, accessStatus string) string {
|
|
if actualSampleCount > 0 {
|
|
return "sampled"
|
|
}
|
|
switch accessStatus {
|
|
case "not_logged_in":
|
|
return "not_logged_in"
|
|
case "unavailable":
|
|
return "unavailable"
|
|
}
|
|
if plannedSampleCount > 0 {
|
|
return "pending"
|
|
}
|
|
if accessStatus == "accessible" {
|
|
return "pending"
|
|
}
|
|
return "pending"
|
|
}
|
|
|
|
func nullableFloat64(value *float64) interface{} {
|
|
if value == nil {
|
|
return nil
|
|
}
|
|
return *value
|
|
}
|
|
|
|
func nullableTime(value sql.NullTime) interface{} {
|
|
if !value.Valid {
|
|
return nil
|
|
}
|
|
return value.Time
|
|
}
|