Expose rule false positive ratio to rule hit table.

The false positive ratio must be computed from the latest outcome
of closed cases where a rule appears, this required adding the org ID
to case events, ingesting some of the case events, and performing a
separate query.

Author: apognu

models/analytics/results.go

@@ -23,11 +23,18 @@ type DecisionsScoreDistribution struct {
 }
 
 type RuleHitTable struct {
-	RuleName       string  `json:"rule_name"`
-	HitCount       int     `json:"hit_count"`
-	HitRatio       float64 `json:"hit_ratio"`
-	DistinctPivots int     `json:"distinct_pivots"`
-	RepeatRatio    float64 `json:"repeat_ratio"`
+	RuleId             uuid.UUID `json:"-"`
+	RuleName           string    `json:"rule_name"`
+	HitCount           int       `json:"hit_count"`
+	HitRatio           float64   `json:"hit_ratio"`
+	FalsePositiveRatio float64   `json:"false_positive_ratio"`
+	DistinctPivots     int       `json:"distinct_pivots"`
+	RepeatRatio        float64   `json:"repeat_ratio"`
+}
+
+type FalsePositiveRatio struct {
+	RuleId             uuid.UUID
+	FalsePositiveRatio float64
 }
 
 type RuleVsDecisionOutcome struct {

models/case_event.go

@@ -3,11 +3,13 @@ package models
 import (
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/guregu/null/v5"
 )
 
 type CaseEvent struct {
 	Id             string
+	OrgId          uuid.UUID
 	CaseId         string
 	UserId         null.String
 	CreatedAt      time.Time
@@ -59,6 +61,7 @@ const (
 )
 
 type CreateCaseEventAttributes struct {
+	OrgId          uuid.UUID
 	CaseId         string
 	UserId         *string
 	EventType      CaseEventType

repositories/analytics_export_repository.go

@@ -10,6 +10,7 @@ import (
 	"github.com/Masterminds/squirrel"
 	"github.com/checkmarble/marble-backend/models"
 	"github.com/checkmarble/marble-backend/models/analytics"
+	"github.com/checkmarble/marble-backend/repositories/dbmodels"
 	"github.com/cockroachdb/errors"
 	"github.com/google/uuid"
 )
@@ -263,6 +264,88 @@ func AnalyticsCopyScreenings(ctx context.Context, exec AnalyticsExecutor, req An
 	return int(nRows), nil
 }
 
+func AnalyticsCopyCaseEvents(ctx context.Context, exec AnalyticsExecutor, req AnalyticsCopyRequest) (int, error) {
+	cte := WithCtesRaw("q", func(b squirrel.StatementBuilderType) squirrel.SelectBuilder {
+		q := b.Select(
+			"ce.id",
+			"ce.org_id",
+			"d.scenario_id",
+			"dr.rule_id as rule_id",
+			"ce.case_id as case_id",
+			"ce.new_value as outcome",
+			"ce.created_at as created_at",
+			"d.trigger_object_type",
+			"row_number() over (partition by ce.case_id order by ce.created_at desc) rnk",
+		).
+			From(dbmodels.TABLE_CASE_EVENTS+" ce").
+			InnerJoin(dbmodels.TABLE_CASES+" c on c.id = ce.case_id").
+			InnerJoin(dbmodels.TABLE_DECISIONS+" d on d.case_id = c.id").
+			InnerJoin(dbmodels.TABLE_DECISION_RULES+" dr on dr.decision_id = d.id").
+			Where("ce.org_id = ?", req.OrgId).
+			Where("ce.event_type = 'outcome_updated'").
+			Where("c.status = 'closed'").
+			Where("d.trigger_object_type = ?", req.TriggerObject).
+			Where("ce.created_at < ?", req.EndTime).
+			OrderBy("ce.created_at, ce.id").
+			Limit(uint64(req.Limit))
+
+		if req.Watermark != nil {
+			q = q.Where("(ce.created_at, ce.id) > (?::timestamp with time zone, ?)",
+				req.Watermark.WatermarkTime, req.Watermark.WatermarkId)
+		}
+
+		return q
+	})
+
+	inner := squirrel.
+		Select(
+			"q.id",
+			"q.scenario_id",
+			"q.rule_id",
+			"q.case_id",
+			"q.outcome",
+			"q.created_at",
+			"q.org_id",
+			"extract(year from q.created_at)::int as year",
+			"extract(month from q.created_at)::int as month",
+			"q.trigger_object_type",
+		).
+		From("q").
+		PrefixExpr(cte).
+		InnerJoin(dbmodels.TABLE_DECISIONS + " d on d.case_id = q.case_id").
+		Where("q.rnk = 1")
+
+	for _, f := range req.TriggerObjectFields {
+		inner = analyticsAddTriggerObjectField(inner, f, false)
+	}
+	for _, f := range req.ExtraDbFields {
+		inner = analyticsAddExtraField(inner, f, false)
+	}
+
+	innerSql, args, err := inner.ToSql()
+	if err != nil {
+		return 0, err
+	}
+
+	unsafeQuery, err := unsafeBuildSqlQuery(innerSql, args)
+	if err != nil {
+		return 0, err
+	}
+
+	query := fmt.Sprintf(`copy ( select * from postgres_query(?, ?) ) to '%s' (format parquet, compression zstd, partition_by (org_id, year, month, trigger_object_type), append)`, req.Table)
+
+	result, err := exec.ExecContext(ctx, query, "pg", unsafeQuery)
+	if err != nil {
+		return 0, err
+	}
+	nRows, err := result.RowsAffected()
+	if err != nil {
+		return 0, err
+	}
+
+	return int(nRows), nil
+}
+
 func analyticsAddTriggerObjectField(b squirrel.SelectBuilder, field models.Field, anyValue bool) squirrel.SelectBuilder {
 	sqlType := "text"
 

repositories/case_event_repository.go

@@ -86,6 +86,7 @@ func (repo *MarbleDbRepository) BatchCreateCaseEvents(ctx context.Context, exec
 
 	query := NewQueryBuilder().Insert(dbmodels.TABLE_CASE_EVENTS).
 		Columns(
+			"org_id",
 			"case_id",
 			"user_id",
 			"event_type",
@@ -105,6 +106,7 @@ func (repo *MarbleDbRepository) BatchCreateCaseEvents(ctx context.Context, exec
 			userId = pgtype.Text{Valid: false}
 		}
 		query = query.Values(
+			createCaseEventAttribute.OrgId,
 			createCaseEventAttribute.CaseId,
 			userId,
 			createCaseEventAttribute.EventType,

repositories/dbmodels/db_case_event.go

@@ -5,11 +5,13 @@ import (
 
 	"github.com/checkmarble/marble-backend/models"
 	"github.com/checkmarble/marble-backend/utils"
+	"github.com/google/uuid"
 	"github.com/guregu/null/v5"
 )
 
 type DBCaseEvent struct {
 	Id             string      `db:"id"`
+	OrgId          uuid.UUID   `db:"org_id"`
 	CaseId         string      `db:"case_id"`
 	UserId         null.String `db:"user_id"`
 	CreatedAt      time.Time   `db:"created_at"`
@@ -44,6 +46,7 @@ func AdaptCaseEvent(caseEvent DBCaseEvent) (models.CaseEvent, error) {
 	}
 	return models.CaseEvent{
 		Id:             caseEvent.Id,
+		OrgId:          caseEvent.OrgId,
 		CaseId:         caseEvent.CaseId,
 		UserId:         caseEvent.UserId,
 		CreatedAt:      caseEvent.CreatedAt,

repositories/migrations/20251118115800_add_case_event_index.sql

@@ -0,0 +1,25 @@
+-- +goose Up
+-- +goose NO TRANSACTION
+-- +goose StatementBegin
+
+alter table case_events
+    add column org_id uuid;
+
+update case_events
+set org_id = c.org_id
+from cases c
+where c.id = case_events.case_id;
+
+alter table case_events
+    alter column org_id set not null;
+
+-- +goose StatementEnd
+
+create index concurrently if not exists idx_case_events_by_org
+    on case_events (org_id, event_type, created_at)
+    where event_type in ('outcome_updated');
+
+-- +goose Down
+
+drop index idx_case_events_by_org;
+alter table case_events drop column org_id;

repositories/sql_utils.go

@@ -13,19 +13,32 @@ type cte struct {
 }
 
 type QueryCte struct {
+	builder func() squirrel.StatementBuilderType
 	queries []cte
 }
 
 func WithCtes(name string, cb func(b squirrel.StatementBuilderType) squirrel.SelectBuilder) *QueryCte {
-	ctes := &QueryCte{}
+	ctes := &QueryCte{
+		builder: NewQueryBuilder,
+	}
+
+	return ctes.With(name, cb)
+}
+
+func WithCtesRaw(name string, cb func(b squirrel.StatementBuilderType) squirrel.SelectBuilder) *QueryCte {
+	ctes := &QueryCte{
+		builder: func() squirrel.StatementBuilderType {
+			return squirrel.StatementBuilder.PlaceholderFormat(squirrel.Question)
+		},
+	}
 
 	return ctes.With(name, cb)
 }
 
 func (q *QueryCte) With(name string, cb func(b squirrel.StatementBuilderType) squirrel.SelectBuilder) *QueryCte {
 	q.queries = append(q.queries, cte{
 		name:  pgx.Identifier.Sanitize([]string{name}),
-		query: cb(NewQueryBuilder()),
+		query: cb(q.builder()),
 	})
 
 	return q

usecases/analytics_query_usecase.go

@@ -97,9 +97,11 @@ func (uc AnalyticsQueryUsecase) RuleHitTable(ctx context.Context, filters dto.An
 
 	query := squirrel.
 		Select(
+			"rule_id",
 			"rule_name",
 			"count() filter (outcome = 'hit') as hit_count",
 			"((count() filter (outcome = 'hit')) / count()) * 100 as hit_ratio",
+			"0 as false_positive_ratio", // Determined in a separate query below
 			"count(distinct pivot_value) filter (outcome = 'hit') as distinct_pivots",
 			"IFNULL(100 - (count(distinct pivot_value) filter (outcome = 'hit') / NULLIF(count() filter (outcome = 'hit'), 0) ) * 100, 0) as repeat_ratio",
 		).
@@ -114,7 +116,60 @@ func (uc AnalyticsQueryUsecase) RuleHitTable(ctx context.Context, filters dto.An
 		return nil, err
 	}
 
-	return repositories.AnalyticsScanStruct[analytics.RuleHitTable](ctx, exec, query)
+	ruleHits, err := repositories.AnalyticsScanStruct[analytics.RuleHitTable](ctx, exec, query)
+	if err != nil {
+		return nil, err
+	}
+
+	// Fetch false positives from another query.
+	//
+	// In case of error (which could happen, for example, because of the
+	// sepatate ingestion between decision rules and case events, return the
+	// initial query, which will have a ratio of false positive of 0).
+	{
+		var cteErr error
+
+		cte := repositories.WithCtes("data", func(b squirrel.StatementBuilderType) squirrel.SelectBuilder {
+			q := b.Select("*").
+				From(uc.analyticsFactory.BuildTarget("rule_hit_outcomes")).
+				Suffix("qualify row_number() over (partition by case_id order by created_at desc) = 1")
+
+			q, cteErr = uc.analyticsFactory.ApplyFilters(q, scenario, filters)
+			if err != nil {
+				return q
+			}
+
+			return q
+		})
+
+		if cteErr != nil {
+			return ruleHits, nil
+		}
+
+		query := squirrel.
+			Select(
+				"rule_id",
+				"ifnull(count() filter (where outcome = 'false_positive') / nullif(count(), 0) * 100, 0) as false_positive_ratio",
+			).
+			PrefixExpr(cte).
+			From("data").
+			GroupBy("rule_id")
+
+		falsePositives, err := repositories.AnalyticsScanStruct[analytics.FalsePositiveRatio](ctx, exec, query)
+		if err != nil {
+			return ruleHits, nil
+		}
+
+		for _, ratio := range falsePositives {
+			for idx, rule := range ruleHits {
+				if ratio.RuleId == rule.RuleId {
+					ruleHits[idx].FalsePositiveRatio = ratio.FalsePositiveRatio
+				}
+			}
+		}
+	}
+
+	return ruleHits, nil
 }
 
 // TODO: could maybe be optimized by storing (d.outcome) denormalized alongside the decision rule.

usecases/auto_assignment_usecase.go

@@ -98,6 +98,7 @@ func (uc AutoAssignmentUsecase) assignCase(ctx context.Context, c models.Case, u
 		}
 
 		if err := uc.caseRepository.CreateCaseEvent(ctx, tx, models.CreateCaseEventAttributes{
+			OrgId:     uuid.MustParse(c.OrganizationId),
 			CaseId:    c.Id,
 			UserId:    nil,
 			EventType: models.CaseAssigned,