From 88d8a2ba8ac0ccf4d1836bc087b2d5bf52459932 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Thu, 3 Sep 2026 20:19:16 -0700 Subject: [PATCH] feat: page every contact timeline source on a row comparison against the (at, source, id) cursor, unnest campaign progress stamps to one row per event so the limit and the logged-open and logged-click suppression apply per event, fetch one row past the page from every source, return all matching suppression entries, and encode the next cursor from the last event's key --- internal/app/contact/handler.go | 4 +- internal/app/contact/service.go | 2 +- internal/repository/pg_contact.go | 382 ++++++++++++++++++------------ 3 files changed, 233 insertions(+), 155 deletions(-) diff --git a/internal/app/contact/handler.go b/internal/app/contact/handler.go index 053967f34..fbbb644d1 100644 --- a/internal/app/contact/handler.go +++ b/internal/app/contact/handler.go @@ -218,6 +218,6 @@ func (s *contactService) ListSentEmails(ctx context.Context, userID, contactID u return s.contactRepository.ListSentEmails(ctx, userID, contactID, limit, beforeSentAt, beforeTaskID) } -func (s *contactService) ListTimeline(ctx context.Context, userID uuid.UUID, orgID *uuid.UUID, contactID uuid.UUID, limit int, before *time.Time) (*models.ContactTimelineResult, *errx.Error) { - return s.contactRepository.ListTimeline(ctx, userID, orgID, contactID, limit, before) +func (s *contactService) ListTimeline(ctx context.Context, userID uuid.UUID, orgID *uuid.UUID, contactID uuid.UUID, limit int, cursor *models.ContactTimelineKey) (*models.ContactTimelineResult, *errx.Error) { + return s.contactRepository.ListTimeline(ctx, userID, orgID, contactID, limit, cursor) } diff --git a/internal/app/contact/service.go b/internal/app/contact/service.go index 6cc28ba8b..88c5a06a6 100644 --- a/internal/app/contact/service.go +++ b/internal/app/contact/service.go @@ -67,7 +67,7 @@ type ContactService interface { // ListTimeline returns a merged, reverse-chronological feed of all // engagement + CRM events for the contact. - ListTimeline(ctx context.Context, userID uuid.UUID, orgID *uuid.UUID, contactID uuid.UUID, limit int, before *time.Time) (*models.ContactTimelineResult, *errx.Error) + ListTimeline(ctx context.Context, userID uuid.UUID, orgID *uuid.UUID, contactID uuid.UUID, limit int, cursor *models.ContactTimelineKey) (*models.ContactTimelineResult, *errx.Error) // SetCampaignWaker wires the campaign service so attaching a lead to a // running campaign wakes that campaign's parked send chain. Optional: with diff --git a/internal/repository/pg_contact.go b/internal/repository/pg_contact.go index c8fcca756..328e33930 100644 --- a/internal/repository/pg_contact.go +++ b/internal/repository/pg_contact.go @@ -92,7 +92,7 @@ type ContactRepository interface { // + deliverability + reply joins are skipped (they're org-scoped). GetDetail(ctx context.Context, userID uuid.UUID, orgID *uuid.UUID, contactID uuid.UUID) (*models.ContactDetail, *errx.Error) ListSentEmails(ctx context.Context, userID, contactID uuid.UUID, limit int, beforeSentAt *time.Time, beforeTaskID *uuid.UUID) (*models.ContactSentEmailsResult, *errx.Error) - ListTimeline(ctx context.Context, userID uuid.UUID, orgID *uuid.UUID, contactID uuid.UUID, limit int, before *time.Time) (*models.ContactTimelineResult, *errx.Error) + ListTimeline(ctx context.Context, userID uuid.UUID, orgID *uuid.UUID, contactID uuid.UUID, limit int, cursor *models.ContactTimelineKey) (*models.ContactTimelineResult, *errx.Error) // ListCampaignStates returns the contact's campaigns with their flow, // this contact's progress on every step, and the derived lead status. ListCampaignStates(ctx context.Context, orgID, contactID uuid.UUID) ([]models.ContactCampaignState, *errx.Error) @@ -2960,11 +2960,21 @@ func (r *contactRepository) ListSentEmails(ctx context.Context, userID, contactI }, nil } +// timelineKeyset is the predicate that pages one source of the contact +// timeline: rows whose (time, source rank, id) tuple sorts strictly before +// the cursor, which the query receives as three consecutive parameters +// starting at $first. Every source uses it so SQL and the merged sort agree. +func timelineKeyset(atCol string, source models.ContactTimelineSource, idCol string, first int) string { + return fmt.Sprintf("(%s, %d, %s) < ($%d::timestamptz, $%d::int, $%d::uuid)", + atCol, source, idCol, first, first+1, first+2) +} + // ListTimeline merges per-contact events from several source tables // into a single, reverse-chronological feed. // // Sources: // - campaign_contact_progress → sent / opened / clicked / replied / bounced +// - email_link_clicks, email_opens → per-event clicks and opens // - reply_intents → received replies (with intent classification) // - deliverability_events → bounce / complaint // - suppressed_recipients → suppression added @@ -2973,18 +2983,21 @@ func (r *contactRepository) ListSentEmails(ctx context.Context, userID, contactI // - contact_activities → creation and campaign / category membership // - website_page_hits → page views from the tracking snippet // -// We pull up to (limit) candidates from each source ordered by time -// DESC, then merge-sort in Go. This avoids a 5-way UNION with -// matching column lists (each source has a different shape), and the -// per-source limit caps the read at roughly 5*limit rows. +// We pull one row past the page from each source, newest first, then +// merge-sort in Go. This avoids a 10-way UNION with matching column lists +// (each source has a different shape), and the per-source limit caps the +// read at roughly 10*limit rows. The lookahead row is what makes has_more +// right when a single source fills the page on its own. // -// The `before` cursor is a wall-clock time; everything strictly older -// than it is eligible. The caller paginates by setting `before` to -// the oldest returned event's `At` on the next call. -func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, orgID *uuid.UUID, contactID uuid.UUID, limit int, before *time.Time) (*models.ContactTimelineResult, *errx.Error) { +// The feed is ordered by (at, source, id) and a page resumes strictly after +// the cursor on that tuple, so two events at the same instant, from the +// same table or different ones, land on one side of a page boundary or the +// other and are never skipped or repeated. A nil cursor is the first page. +func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, orgID *uuid.UUID, contactID uuid.UUID, limit int, cursor *models.ContactTimelineKey) (*models.ContactTimelineResult, *errx.Error) { if limit <= 0 || limit > 200 { limit = 50 } + fetch := limit + 1 // We resolve the contact's email up front because some org-scoped // joins (suppression, deliverability fallback, reply_intents) key @@ -3001,42 +3014,42 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, return nil, errx.InternalError() } - // "before" defaults to "now + 1 minute" so the first page picks - // up everything. Using a future bound keeps the SQL uniform — every - // query passes the same predicate. - bound := time.Now().Add(time.Minute) - if before != nil { - bound = *before + // The position the page resumes after. The first page starts a minute + // in the future at the lowest rank, which admits every event the same + // way a real cursor would, so every query passes the same predicate. + after := models.ContactTimelineKey{At: time.Now().Add(time.Minute)} + if cursor != nil { + after = *cursor } + afterSource := int(after.Source) events := make([]models.ContactTimelineEvent, 0, limit*2) - // 1. Engagement events from campaign_contact_progress. One progress - // row can emit up to 5 events (sent/opened/clicked/replied/bounced). - // The coarse clicked stamp is emitted only when no logged click stands - // for it (a stamp written before per-link logging); otherwise source 9 - // names the link. A logged click represents the stamp when it landed - // within a minute of it, the stamp being written as the click is logged. - progressQuery := ` + // 1. Engagement stamps from campaign_contact_progress, unnested to one + // row per stamp so the limit and the cursor apply to events, not to + // leads: a lead whose newest stamp is past the cursor must not push + // an older lead's eligible stamp off the page. The coarse opened and + // clicked stamps are emitted only when no logged open or click stands + // for them (a stamp written before per-event logging); otherwise + // sources 9 and 10 carry the event with its origin. A logged event + // represents the stamp when it landed within a minute of it, the + // stamp being written as the event is logged. + progressQuery := fmt.Sprintf(` SELECT - ccp.sent_at, ccp.opened_at, ccp.clicked_at, ccp.replied_at, ccp.bounced_at, - ccp.opened_machine, - (ccp.opened_at IS NOT NULL AND EXISTS ( - SELECT 1 FROM email_opens o - WHERE o.campaign_id = ccp.campaign_id AND o.contact_id = ccp.contact_id AND o.sequence_id = ccp.sequence_id - AND o.opened_at BETWEEN ccp.opened_at - INTERVAL '1 minute' AND ccp.opened_at + INTERVAL '1 minute' - )) AS has_open_log, - (ccp.clicked_at IS NOT NULL AND EXISTS ( - SELECT 1 FROM email_link_clicks lc - WHERE lc.campaign_id = ccp.campaign_id AND lc.contact_id = ccp.contact_id AND lc.sequence_id = ccp.sequence_id - AND lc.clicked_at BETWEEN ccp.clicked_at - INTERVAL '1 minute' AND ccp.clicked_at + INTERVAL '1 minute' - )) AS has_link_clicks, + ev.source, ev.at, ccp.sequence_id, ccp.opened_machine, cam.id, cam.name, seq.id, seq.name, seq.subject, ea.id, ea.email, ea.name FROM campaign_contact_progress ccp JOIN campaigns cam ON cam.id = ccp.campaign_id JOIN sequences seq ON seq.id = ccp.sequence_id + CROSS JOIN LATERAL (VALUES + (%[1]d, ccp.sent_at), + (%[2]d, ccp.opened_at), + (%[3]d, ccp.clicked_at), + (%[4]d, ccp.replied_at), + (%[5]d, ccp.bounced_at) + ) AS ev(source, at) LEFT JOIN LATERAL ( SELECT ea.id, ea.email, ea.name FROM tasks t @@ -3050,29 +3063,41 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, ) ea ON TRUE WHERE ccp.contact_id = $1 AND cam.user_id = $2 - AND COALESCE(ccp.sent_at, ccp.opened_at, ccp.clicked_at, ccp.replied_at, ccp.bounced_at) < $3 - ORDER BY GREATEST( - COALESCE(ccp.sent_at, 'epoch'), - COALESCE(ccp.opened_at, 'epoch'), - COALESCE(ccp.clicked_at, 'epoch'), - COALESCE(ccp.replied_at, 'epoch'), - COALESCE(ccp.bounced_at, 'epoch') - ) DESC - LIMIT $4 - ` - prows, err := r.DB.Query(ctx, progressQuery, contactID, userID, bound, limit) + AND ev.at IS NOT NULL + AND (ev.at, ev.source, ccp.sequence_id) < ($3::timestamptz, $4::int, $5::uuid) + AND NOT (ev.source = %[2]d AND EXISTS ( + SELECT 1 FROM email_opens o + WHERE o.campaign_id = ccp.campaign_id AND o.contact_id = ccp.contact_id AND o.sequence_id = ccp.sequence_id + AND o.opened_at BETWEEN ccp.opened_at - INTERVAL '1 minute' AND ccp.opened_at + INTERVAL '1 minute' + )) + AND NOT (ev.source = %[3]d AND EXISTS ( + SELECT 1 FROM email_link_clicks lc + WHERE lc.campaign_id = ccp.campaign_id AND lc.contact_id = ccp.contact_id AND lc.sequence_id = ccp.sequence_id + AND lc.clicked_at BETWEEN ccp.clicked_at - INTERVAL '1 minute' AND ccp.clicked_at + INTERVAL '1 minute' + )) + ORDER BY ev.at DESC, ev.source DESC, ccp.sequence_id DESC + LIMIT $6 + `, + models.TimelineSourceProgressSent, + models.TimelineSourceProgressOpened, + models.TimelineSourceProgressClicked, + models.TimelineSourceProgressReplied, + models.TimelineSourceProgressBounced, + ) + prows, err := r.DB.Query(ctx, progressQuery, contactID, userID, after.At, afterSource, after.ID, fetch) if err != nil { - db.CaptureError(err, progressQuery, []any{contactID, userID, bound, limit}, "ListTimeline progress") + db.CaptureError(err, progressQuery, []any{contactID, userID, after.At, afterSource, after.ID, fetch}, "ListTimeline progress") return nil, errx.InternalError() } for prows.Next() { - var sentAt, openedAt, clickedAt, repliedAt, bouncedAt *time.Time - var openedMachine, hasOpenLog, hasLinkClicks bool + var source int + var at time.Time + var seqKey uuid.UUID + var openedMachine bool var campID, seqID, eaID *uuid.UUID var campName, seqName, seqSubject, eaEmail, eaName *string if err := prows.Scan( - &sentAt, &openedAt, &clickedAt, &repliedAt, &bouncedAt, - &openedMachine, &hasOpenLog, &hasLinkClicks, + &source, &at, &seqKey, &openedMachine, &campID, &campName, &seqID, &seqName, &seqSubject, &eaID, &eaEmail, &eaName, @@ -3081,46 +3106,43 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, db.CaptureError(err, "", nil, "ListTimeline progress scan") return nil, errx.InternalError() } - baseSubject := seqSubject - makeEvent := func(t *time.Time, ty models.ContactTimelineEventType) { - if t == nil || !t.Before(bound) { - return - } - ev := models.ContactTimelineEvent{ - Type: ty, - At: *t, - EmailAccountID: eaID, - EmailAccountEmail: eaEmail, - EmailAccountName: eaName, - CampaignID: campID, - CampaignName: campName, - SequenceID: seqID, - SequenceName: seqName, - } - if baseSubject != nil && *baseSubject != "" { - ev.Subject = baseSubject - } - if ty == models.TimelineEmailOpened { - machine := openedMachine - ev.Machine = &machine - } - events = append(events, ev) + ev := models.ContactTimelineEvent{ + At: at, + Key: models.ContactTimelineKey{At: at, Source: models.ContactTimelineSource(source), ID: seqKey}, + EmailAccountID: eaID, + EmailAccountEmail: eaEmail, + EmailAccountName: eaName, + CampaignID: campID, + CampaignName: campName, + SequenceID: seqID, + SequenceName: seqName, } - makeEvent(sentAt, models.TimelineEmailSent) - // A step whose first open is in the log hands its opens to source 10, - // one row per open with its origin; the summary column only stands in - // for a first open the log never saw (a step tracked before the log - // existed), even when later opens were logged. - if !hasOpenLog { - makeEvent(openedAt, models.TimelineEmailOpened) + if seqSubject != nil && *seqSubject != "" { + ev.Subject = seqSubject } - if !hasLinkClicks { - makeEvent(clickedAt, models.TimelineEmailClicked) + switch ev.Key.Source { + case models.TimelineSourceProgressSent: + ev.Type = models.TimelineEmailSent + case models.TimelineSourceProgressOpened: + ev.Type = models.TimelineEmailOpened + machine := openedMachine + ev.Machine = &machine + case models.TimelineSourceProgressClicked: + ev.Type = models.TimelineEmailClicked + case models.TimelineSourceProgressReplied: + ev.Type = models.TimelineEmailReplied + case models.TimelineSourceProgressBounced: + ev.Type = models.TimelineEmailBounced + default: + continue } - makeEvent(repliedAt, models.TimelineEmailReplied) - makeEvent(bouncedAt, models.TimelineEmailBounced) + events = append(events, ev) } prows.Close() + if err := prows.Err(); err != nil { + db.CaptureError(err, progressQuery, nil, "ListTimeline progress rows") + return nil, errx.InternalError() + } // 9. Per-link clicks: which link, where it went, and whether a person or // a scanner clicked it. Same campaign scope as the progress feed. @@ -3141,13 +3163,13 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, ) ea ON TRUE WHERE lc.contact_id = $1 AND cam.user_id = $2 - AND lc.clicked_at < $3 - ORDER BY lc.clicked_at DESC - LIMIT $4 + AND ` + timelineKeyset("lc.clicked_at", models.TimelineSourceLinkClick, "lc.id", 3) + ` + ORDER BY lc.clicked_at DESC, lc.id DESC + LIMIT $6 ` - crows, err := r.DB.Query(ctx, clickQuery, contactID, userID, bound, limit+1) + crows, err := r.DB.Query(ctx, clickQuery, contactID, userID, after.At, afterSource, after.ID, fetch) if err != nil { - db.CaptureError(err, clickQuery, []any{contactID, userID, bound, limit}, "ListTimeline link clicks") + db.CaptureError(err, clickQuery, []any{contactID, userID, after.At, afterSource, after.ID, fetch}, "ListTimeline link clicks") return nil, errx.InternalError() } for crows.Next() { @@ -3175,6 +3197,7 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, ev := models.ContactTimelineEvent{ Type: models.TimelineEmailClicked, At: at, + Key: models.ContactTimelineKey{At: at, Source: models.TimelineSourceLinkClick, ID: link.ID}, EmailAccountID: eaID, EmailAccountEmail: eaEmail, EmailAccountName: eaName, @@ -3224,13 +3247,13 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, ) ea ON TRUE WHERE o.contact_id = $1 AND cam.user_id = $2 - AND o.opened_at < $3 - ORDER BY o.opened_at DESC - LIMIT $4 + AND ` + timelineKeyset("o.opened_at", models.TimelineSourceOpen, "o.id", 3) + ` + ORDER BY o.opened_at DESC, o.id DESC + LIMIT $6 ` - orows, err := r.DB.Query(ctx, openQuery, contactID, userID, bound, limit+1) + orows, err := r.DB.Query(ctx, openQuery, contactID, userID, after.At, afterSource, after.ID, fetch) if err != nil { - db.CaptureError(err, openQuery, []any{contactID, userID, bound, limit}, "ListTimeline opens") + db.CaptureError(err, openQuery, []any{contactID, userID, after.At, afterSource, after.ID, fetch}, "ListTimeline opens") return nil, errx.InternalError() } for orows.Next() { @@ -3256,6 +3279,7 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, ev := models.ContactTimelineEvent{ Type: models.TimelineEmailOpened, At: at, + Key: models.ContactTimelineKey{At: at, Source: models.TimelineSourceOpen, ID: id}, EmailAccountID: eaID, EmailAccountEmail: eaEmail, EmailAccountName: eaName, @@ -3288,60 +3312,68 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, if orgID != nil { // 2. Reply intents (inbound replies with classification). replyQuery := ` - SELECT ri.created_at, ri.intent, ri.campaign_id, cam.name, ri.task_id + SELECT ri.id, ri.created_at, ri.intent, ri.campaign_id, cam.name, ri.task_id FROM reply_intents ri LEFT JOIN campaigns cam ON cam.id = ri.campaign_id WHERE ri.organization_id = $1 AND LOWER(ri.contact_email) = LOWER($2) - AND ri.created_at < $3 - ORDER BY ri.created_at DESC - LIMIT $4 + AND ` + timelineKeyset("ri.created_at", models.TimelineSourceReplyIntent, "ri.id", 3) + ` + ORDER BY ri.created_at DESC, ri.id DESC + LIMIT $6 ` - rrows, err := r.DB.Query(ctx, replyQuery, *orgID, contactEmail, bound, limit) + rrows, err := r.DB.Query(ctx, replyQuery, *orgID, contactEmail, after.At, afterSource, after.ID, fetch) if err != nil { db.CaptureError(err, replyQuery, nil, "ListTimeline replies") return nil, errx.InternalError() } for rrows.Next() { var ev models.ContactTimelineEvent + var id uuid.UUID var intent string - if err := rrows.Scan(&ev.At, &intent, &ev.CampaignID, &ev.CampaignName, &ev.TaskID); err != nil { + if err := rrows.Scan(&id, &ev.At, &intent, &ev.CampaignID, &ev.CampaignName, &ev.TaskID); err != nil { rrows.Close() db.CaptureError(err, "", nil, "ListTimeline replies scan") return nil, errx.InternalError() } ev.Type = models.TimelineReplyReceived + ev.Key = models.ContactTimelineKey{At: ev.At, Source: models.TimelineSourceReplyIntent, ID: id} ev.Intent = &intent events = append(events, ev) } rrows.Close() + if err := rrows.Err(); err != nil { + db.CaptureError(err, replyQuery, nil, "ListTimeline replies rows") + return nil, errx.InternalError() + } // 3. Deliverability events (bounce / complaint / unsubscribe). delivQuery := ` - SELECT de.created_at, de.event_type, de.provider, de.reason, + SELECT de.id, de.created_at, de.event_type, de.provider, de.reason, de.campaign_id, cam.name, de.task_id FROM deliverability_events de LEFT JOIN campaigns cam ON cam.id = de.campaign_id WHERE de.organization_id = $1 AND (de.contact_id = $2 OR LOWER(de.recipient_email) = LOWER($3)) - AND de.created_at < $4 - ORDER BY de.created_at DESC - LIMIT $5 + AND ` + timelineKeyset("de.created_at", models.TimelineSourceDeliverability, "de.id", 4) + ` + ORDER BY de.created_at DESC, de.id DESC + LIMIT $7 ` - drows, err := r.DB.Query(ctx, delivQuery, *orgID, contactID, contactEmail, bound, limit) + drows, err := r.DB.Query(ctx, delivQuery, *orgID, contactID, contactEmail, after.At, afterSource, after.ID, fetch) if err != nil { db.CaptureError(err, delivQuery, nil, "ListTimeline deliv") return nil, errx.InternalError() } for drows.Next() { var ev models.ContactTimelineEvent + var id uuid.UUID var eventType, provider, reason string - if err := drows.Scan(&ev.At, &eventType, &provider, &reason, &ev.CampaignID, &ev.CampaignName, &ev.TaskID); err != nil { + if err := drows.Scan(&id, &ev.At, &eventType, &provider, &reason, &ev.CampaignID, &ev.CampaignName, &ev.TaskID); err != nil { drows.Close() db.CaptureError(err, "", nil, "ListTimeline deliv scan") return nil, errx.InternalError() } ev.Type = models.TimelineDeliverability + ev.Key = models.ContactTimelineKey{At: ev.At, Source: models.TimelineSourceDeliverability, ID: id} ev.Source = &eventType ev.Provider = &provider if reason != "" { @@ -3350,93 +3382,118 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, events = append(events, ev) } drows.Close() + if err := drows.Err(); err != nil { + db.CaptureError(err, delivQuery, nil, "ListTimeline deliv rows") + return nil, errx.InternalError() + } - // 4. Suppression — emit one event at create time. We treat - // later updates as the same event for now. + // 4. Suppression: one event per matching entry (the address itself + // and its domain), at create time. Later updates are the same event. suppQuery := ` - SELECT created_at, reason, source + SELECT id, created_at, reason, source FROM suppressed_recipients WHERE organization_id = $1 AND ((kind = 'email' AND email = LOWER($2)) OR (kind = 'domain' AND email = split_part(LOWER($2), '@', 2))) - AND created_at < $3 - ORDER BY created_at DESC - LIMIT 1 + AND ` + timelineKeyset("created_at", models.TimelineSourceSuppression, "id", 3) + ` + ORDER BY created_at DESC, id DESC + LIMIT $6 ` - var sAt time.Time - var sReason, sSource string - if err := r.DB.QueryRow(ctx, suppQuery, *orgID, contactEmail, bound).Scan(&sAt, &sReason, &sSource); err == nil { + srows, err := r.DB.Query(ctx, suppQuery, *orgID, contactEmail, after.At, afterSource, after.ID, fetch) + if err != nil { + db.CaptureError(err, suppQuery, nil, "ListTimeline suppression") + return nil, errx.InternalError() + } + for srows.Next() { + var id uuid.UUID + var sAt time.Time + var sReason, sSource string + if err := srows.Scan(&id, &sAt, &sReason, &sSource); err != nil { + srows.Close() + db.CaptureError(err, "", nil, "ListTimeline suppression scan") + return nil, errx.InternalError() + } ev := models.ContactTimelineEvent{ Type: models.TimelineSuppressed, At: sAt, + Key: models.ContactTimelineKey{At: sAt, Source: models.TimelineSourceSuppression, ID: id}, Source: &sSource, } if sReason != "" { ev.Reason = &sReason } events = append(events, ev) - } else if err != pgx.ErrNoRows { - db.CaptureError(err, suppQuery, nil, "ListTimeline suppression") + } + srows.Close() + if err := srows.Err(); err != nil { + db.CaptureError(err, suppQuery, nil, "ListTimeline suppression rows") return nil, errx.InternalError() } // 5. Notes. notesQuery := ` - SELECT created_at, user_id, content + SELECT id, created_at, user_id, content FROM contact_notes WHERE contact_id = $1 AND organization_id = $2 - AND created_at < $3 - ORDER BY created_at DESC - LIMIT $4 + AND ` + timelineKeyset("created_at", models.TimelineSourceNote, "id", 3) + ` + ORDER BY created_at DESC, id DESC + LIMIT $6 ` - nrows, err := r.DB.Query(ctx, notesQuery, contactID, *orgID, bound, limit) + nrows, err := r.DB.Query(ctx, notesQuery, contactID, *orgID, after.At, afterSource, after.ID, fetch) if err != nil { db.CaptureError(err, notesQuery, nil, "ListTimeline notes") return nil, errx.InternalError() } for nrows.Next() { var ev models.ContactTimelineEvent - var uid uuid.UUID + var id, uid uuid.UUID var content string - if err := nrows.Scan(&ev.At, &uid, &content); err != nil { + if err := nrows.Scan(&id, &ev.At, &uid, &content); err != nil { nrows.Close() db.CaptureError(err, "", nil, "ListTimeline notes scan") return nil, errx.InternalError() } ev.Type = models.TimelineNote + ev.Key = models.ContactTimelineKey{At: ev.At, Source: models.TimelineSourceNote, ID: id} ev.UserID = &uid ev.Content = &content events = append(events, ev) } nrows.Close() + if err := nrows.Err(); err != nil { + db.CaptureError(err, notesQuery, nil, "ListTimeline notes rows") + return nil, errx.InternalError() + } // 6. Meetings booked through a connected scheduling provider. The event // time is when the booking arrived; scheduled_for carries the call // window so the UI can render "Meeting on ". meetingQuery := ` - SELECT created_at, status, source, event_name, scheduled_for, join_url, canceled_reason + SELECT id, created_at, status, source, event_name, scheduled_for, join_url, canceled_reason FROM meeting_bookings WHERE contact_id = $1 AND organization_id = $2 - AND created_at < $3 - ORDER BY created_at DESC - LIMIT $4 + AND ` + timelineKeyset("created_at", models.TimelineSourceMeeting, "id", 3) + ` + ORDER BY created_at DESC, id DESC + LIMIT $6 ` - mrows, err := r.DB.Query(ctx, meetingQuery, contactID, *orgID, bound, limit) + mrows, err := r.DB.Query(ctx, meetingQuery, contactID, *orgID, after.At, afterSource, after.ID, fetch) if err != nil { db.CaptureError(err, meetingQuery, nil, "ListTimeline meetings") return nil, errx.InternalError() } for mrows.Next() { var ev models.ContactTimelineEvent + var id uuid.UUID var status, source, eventName, joinURL, canceledReason string var scheduledFor *time.Time - if err := mrows.Scan(&ev.At, &status, &source, &eventName, &scheduledFor, &joinURL, &canceledReason); err != nil { + if err := mrows.Scan(&id, &ev.At, &status, &source, &eventName, &scheduledFor, &joinURL, &canceledReason); err != nil { mrows.Close() db.CaptureError(err, "", nil, "ListTimeline meetings scan") return nil, errx.InternalError() } + ev.Key = models.ContactTimelineKey{At: ev.At, Source: models.TimelineSourceMeeting, ID: id} switch status { case "rescheduled": ev.Type = models.TimelineMeetingRescheduled @@ -3463,36 +3520,42 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, events = append(events, ev) } mrows.Close() + if err := mrows.Err(); err != nil { + db.CaptureError(err, meetingQuery, nil, "ListTimeline meetings rows") + return nil, errx.InternalError() + } // 7. Lifecycle: creation (with its first-touch source) and campaign / // category membership changes, from contact_activities. Names were // resolved when the row was written, so a renamed or deleted // campaign still reads correctly. lifeQuery := ` - SELECT created_at, user_id, activity_type, metadata + SELECT id, created_at, user_id, activity_type, metadata FROM contact_activities WHERE contact_id = $1 AND organization_id = $2 AND activity_type IN ('contact_created', 'campaign_added', 'campaign_removed', 'category_added', 'category_removed', 'form_submitted') - AND created_at < $3 - ORDER BY created_at DESC - LIMIT $4 + AND ` + timelineKeyset("created_at", models.TimelineSourceActivity, "id", 3) + ` + ORDER BY created_at DESC, id DESC + LIMIT $6 ` - lrows, err := r.DB.Query(ctx, lifeQuery, contactID, *orgID, bound, limit) + lrows, err := r.DB.Query(ctx, lifeQuery, contactID, *orgID, after.At, afterSource, after.ID, fetch) if err != nil { db.CaptureError(err, lifeQuery, nil, "ListTimeline lifecycle") return nil, errx.InternalError() } for lrows.Next() { var ev models.ContactTimelineEvent + var rowID uuid.UUID var typ string var meta map[string]any - if err := lrows.Scan(&ev.At, &ev.UserID, &typ, &meta); err != nil { + if err := lrows.Scan(&rowID, &ev.At, &ev.UserID, &typ, &meta); err != nil { lrows.Close() db.CaptureError(err, "", nil, "ListTimeline lifecycle scan") return nil, errx.InternalError() } ev.Type = models.ContactTimelineEventType(typ) + ev.Key = models.ContactTimelineKey{At: ev.At, Source: models.TimelineSourceActivity, ID: rowID} str := func(k string) *string { if v, ok := meta[k].(string); ok && v != "" { return &v @@ -3524,6 +3587,10 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, events = append(events, ev) } lrows.Close() + if err := lrows.Err(); err != nil { + db.CaptureError(err, lifeQuery, nil, "ListTimeline lifecycle rows") + return nil, errx.InternalError() + } // 8. Website page views from any browser tied to the contact through // an email-link ticket. @@ -3537,11 +3604,11 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, FROM website_page_hits h WHERE h.organization_id = $1 AND h.visitor_id IN (SELECT id FROM website_visitors WHERE contact_id = $2) - AND h.occurred_at < $3 - ORDER BY h.occurred_at DESC - LIMIT $4 + AND ` + timelineKeyset("h.occurred_at", models.TimelineSourcePageHit, "h.id", 3) + ` + ORDER BY h.occurred_at DESC, h.id DESC + LIMIT $6 ` - hrows, err := r.DB.Query(ctx, hitQuery, *orgID, contactID, bound, limit) + hrows, err := r.DB.Query(ctx, hitQuery, *orgID, contactID, after.At, afterSource, after.ID, fetch) if err != nil { db.CaptureError(err, hitQuery, nil, "ListTimeline page hits") return nil, errx.InternalError() @@ -3561,7 +3628,12 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, return nil, errx.InternalError() } hit := h - ev := models.ContactTimelineEvent{Type: models.TimelinePageHit, At: h.OccurredAt, PageHit: &hit} + ev := models.ContactTimelineEvent{ + Type: models.TimelinePageHit, + At: h.OccurredAt, + Key: models.ContactTimelineKey{At: h.OccurredAt, Source: models.TimelineSourcePageHit, ID: h.ID}, + PageHit: &hit, + } subject := h.Title if subject == "" { subject = h.Path @@ -3570,21 +3642,27 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, events = append(events, ev) } hrows.Close() + if err := hrows.Err(); err != nil { + db.CaptureError(err, hitQuery, nil, "ListTimeline page hits rows") + return nil, errx.InternalError() + } } - // Merge sort: newest first. - sort.Slice(events, func(i, j int) bool { return events[i].At.After(events[j].At) }) + // Merge sort: newest first, ties broken exactly as each source query + // broke them, so the page boundary is the same position everywhere. + sort.Slice(events, func(i, j int) bool { return events[j].Key.Before(events[i].Key) }) - hasMore := false + res := &models.ContactTimelineResult{Data: events} if len(events) > limit { - hasMore = true - events = events[:limit] + res.Data = events[:limit] + last := res.Data[limit-1].Key + res.HasMore = true + res.Pagination = models.Pagination{ + NextCursor: paging.EncodeMerged(last.At, int(last.Source), last.ID), + HasMore: true, + } } - - return &models.ContactTimelineResult{ - Data: events, - HasMore: hasMore, - }, nil + return res, nil } // fillUTM reads the UTM parameters off a clicked link's destination, whether