@@ -266,7 +266,7 @@ func AnalyticsCopyScreenings(ctx context.Context, exec AnalyticsExecutor, req An
266266
267267func AnalyticsCopyCaseEvents (ctx context.Context , exec AnalyticsExecutor , req AnalyticsCopyRequest ) (int , error ) {
268268 cte := WithCtesRaw ("q" , func (b squirrel.StatementBuilderType ) squirrel.SelectBuilder {
269- q := b .Select (
269+ q1 := b .Select (
270270 "ce.id" ,
271271 "ce.org_id" ,
272272 "d.scenario_id" ,
@@ -276,6 +276,7 @@ func AnalyticsCopyCaseEvents(ctx context.Context, exec AnalyticsExecutor, req An
276276 "ce.created_at as created_at" ,
277277 "d.trigger_object_type" ,
278278 "row_number() over (partition by ce.case_id order by ce.created_at desc) rnk" ,
279+ "dense_rank() over (order by ce.created_at, ce.id) as deduplicate" ,
279280 ).
280281 From (dbmodels .TABLE_CASE_EVENTS + " ce" ).
281282 InnerJoin (dbmodels .TABLE_CASES + " c on c.id = ce.case_id" ).
@@ -290,11 +291,13 @@ func AnalyticsCopyCaseEvents(ctx context.Context, exec AnalyticsExecutor, req An
290291 Limit (uint64 (req .Limit ))
291292
292293 if req .Watermark != nil {
293- q = q .Where ("(ce.created_at, ce.id) > (?::timestamp with time zone, ?)" ,
294+ q1 = q1 .Where ("(ce.created_at, ce.id) > (?::timestamp with time zone, ?)" ,
294295 req .Watermark .WatermarkTime , req .Watermark .WatermarkId )
295296 }
296297
297- return q
298+ return b .
299+ Select ("*" , "max(deduplicate) over() as max_deduplicate" ).
300+ FromSelect (q1 , "i" )
298301 })
299302
300303 inner := squirrel .
@@ -313,7 +316,7 @@ func AnalyticsCopyCaseEvents(ctx context.Context, exec AnalyticsExecutor, req An
313316 From ("q" ).
314317 PrefixExpr (cte ).
315318 InnerJoin (dbmodels .TABLE_DECISIONS + " d on d.case_id = q.case_id" ).
316- Where ("q.rnk = 1" )
319+ Where ("q.rnk = 1 and deduplicate < max_deduplicate " )
317320
318321 for _ , f := range req .TriggerObjectFields {
319322 inner = analyticsAddTriggerObjectField (inner , f , false )
0 commit comments