diff --git a/Makefile b/Makefile index 61200e3b..acad3d11 100644 --- a/Makefile +++ b/Makefile @@ -626,6 +626,7 @@ forms-web: consumer: $(GO_DEV_ENV) \ $(AI_DEV_ENV) \ + GEODB_PATH=data/GeoLite2-City.mmdb \ go run ./cmd/consumer # Send/sync worker. No Postgres by design. WORKER_ID is an explicit UUID diff --git a/cmd/consumer/main.go b/cmd/consumer/main.go index 4deb9fa4..840b4899 100644 --- a/cmd/consumer/main.go +++ b/cmd/consumer/main.go @@ -47,6 +47,7 @@ import ( "github.com/warmbly/warmbly/internal/observability" "github.com/warmbly/warmbly/internal/pkg/encrypt" "github.com/warmbly/warmbly/internal/pkg/generation" + "github.com/warmbly/warmbly/internal/pkg/geo" "github.com/warmbly/warmbly/internal/repository" ) @@ -468,6 +469,14 @@ func main() { // open/click action chains (advancedService), the open/click analog of the // reply path. Decodes with the same codec the Rust tracking service writes // (Avro on Kafka, JSON on NATS). + // GeoIP is optional here as on the backend: it only turns an open or + // click's source network into a country and city on the logs. + geoPath, _ := cfg.LoadGeoDBPath(ctx) + geoloc, gerr := geo.New(geoPath) + if gerr != nil { + log.Printf("GeoIP database not found at %s; engagement locations are disabled.", geoPath) + geoloc, _ = geo.New("") + } if trackingCfg, terr := cfg.LoadTrackingConsumerConfig(ctx); terr != nil { log.Println("tracking consumer config unavailable; opens/clicks not consumed:", terr) } else if trackingConsumer, terr := jobs.NewTrackingConsumer( @@ -485,6 +494,8 @@ func main() { repository.NewLinkClickRepository(primaryDB.Pool), advancedService, verificationEvidence, + repository.NewEmailOpenRepository(primaryDB.Pool), + geoloc, ); terr != nil { log.Println("tracking consumer unavailable; opens/clicks not consumed:", terr) } else { diff --git a/deploy/config/env.example b/deploy/config/env.example index a8e41956..3625b88f 100644 --- a/deploy/config/env.example +++ b/deploy/config/env.example @@ -246,6 +246,10 @@ TRACKING_PAGEHIT_RATE_LIMIT_PER_MIN=60 # The header that proxy sets with the client address (x-forwarded-for, or # cf-connecting-ip behind Cloudflare). Nothing else is read. # TRACKING_CLIENT_IP_HEADER=x-forwarded-for +# Secret the source-address token in tracking events is keyed with; defaults +# to INTERNAL_API_TOKEN. Set it apart to rotate the token without rotating +# the internal API token. +# TRACKING_IP_HASH_KEY= # === Forms service (hosted form pages, embeds, submissions) === # Its own process (cmd/forms) on FORMS_PORT, so form traffic never touches the diff --git a/docker-compose.yml b/docker-compose.yml index 8e926504..a393f301 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -414,6 +414,9 @@ services: environment: <<: *selfhost-env ENCRYPTED_KEYS_PROVIDER: postgres + # Optional, as on the backend: turns an open or click's network into a + # country and city on the contact's timeline. + GEODB_PATH: ${GEODB_PATH:-/app/data/GeoLite2-City.mmdb} volumes: - blobs:/data/blobs depends_on: @@ -480,6 +483,7 @@ services: TRACKING_PAGEHIT_RATE_LIMIT_PER_MIN: ${TRACKING_PAGEHIT_RATE_LIMIT_PER_MIN:-} TRACKING_TRUSTED_PROXIES: ${TRACKING_TRUSTED_PROXIES:-} TRACKING_CLIENT_IP_HEADER: ${TRACKING_CLIENT_IP_HEADER:-} + TRACKING_IP_HASH_KEY: ${TRACKING_IP_HASH_KEY:-} SENTRY_DSN: ${SENTRY_DSN:-} # Overridable so `make dev` (native backend) can point at # host.docker.internal:8080 while `docker compose up` uses the container. diff --git a/docs/content/docs/api/realtime.mdx b/docs/content/docs/api/realtime.mdx index c9977b93..752f95b6 100644 --- a/docs/content/docs/api/realtime.mdx +++ b/docs/content/docs/api/realtime.mdx @@ -41,7 +41,7 @@ After connecting, join one or more topics with a `phx_join` message: | `account:` | One mailbox's sync and warmup events | Requires `manage_emails` | | `bulk:` | Progress of one bulk operation | Import/export progress; only the user who started the operation receives its events | -Events arrive as channel messages whose event name is the event type, for example `EMAIL_SENT`, `EMAIL_OPENED`, `EMAIL_REPLIED`, `EMAIL_RECEIVED`, `AI_DRAFT_READY`, `CAMPAIGN_COMPLETED`, `TASK_PROGRESS`, `ACCOUNT_HEALTH_CHANGED`, `ACCOUNT_SYNC_STATE`, `AUDIT_CREATED`, `AUTOMATION_RUN`, `MEETING_BOOKED`, `NOTIFICATION_CREATED`, `PAGE_HIT`. Payloads always include `event_type` and a timestamp, plus the relevant ids (`campaign_id`, `contact_id`, `thread_id`, and so on). +Events arrive as channel messages whose event name is the event type, for example `EMAIL_SENT`, `EMAIL_OPENED`, `EMAIL_REPLIED`, `EMAIL_RECEIVED`, `AI_DRAFT_READY`, `CAMPAIGN_COMPLETED`, `TASK_PROGRESS`, `ACCOUNT_HEALTH_CHANGED`, `ACCOUNT_SYNC_STATE`, `AUDIT_CREATED`, `AUTOMATION_RUN`, `MEETING_BOOKED`, `NOTIFICATION_CREATED`, `PAGE_HIT`. Payloads always include `event_type` and a timestamp, plus the relevant ids (`campaign_id`, `contact_id`, `thread_id`, and so on). `EMAIL_OPENED` and `EMAIL_CLICKED` also carry `occurred_at` (when the tracking service saw it; `timestamp` is the publish time), `machine` (true for a mail client prefetch or a security scanner's fetch), and `client`, `device_type`, `country_code` and `city` when the consumer could tell. `AI_DRAFT_READY` fires when the [inbox agent](/guides/inbox-agent/) drafts a suggested reply awaiting review; it carries `thread_id` and `draft_id` and requires `access_unibox`. diff --git a/docs/content/docs/api/reference/analytics.mdx b/docs/content/docs/api/reference/analytics.mdx index be4641e2..a09387ab 100644 --- a/docs/content/docs/api/reference/analytics.mdx +++ b/docs/content/docs/api/reference/analytics.mdx @@ -260,10 +260,17 @@ Returns a single campaign's performance summary plus per-sequence-step stats. Th "replies": 28, "bounces": 3 } - ] + ], + "engagement": { + "countries": [{ "key": "US", "opens": 120, "clicks": 31 }, { "key": "DE", "opens": 44, "clicks": 9 }], + "clients": [{ "key": "Gmail", "opens": 98, "clicks": 20 }, { "key": "Outlook", "opens": 51, "clicks": 12 }], + "devices": [{ "key": "desktop", "opens": 140, "clicks": 37 }, { "key": "mobile", "opens": 24, "clicks": 3 }] + } } ``` +`engagement` groups the campaign's human opens and clicks by country (ISO code), mail client or browser, and device type (`desktop`, `mobile`, `tablet`), counting distinct contacts per bucket; a click counts as an open. Each list holds the busiest eight; an empty `key` is unknown. Country needs the GeoLite2 database on the consumer, otherwise every country row is unknown. + `machine_opens` is the subset of `unique_opens` from automated fetchers (Apple MPP prefetch, UA-less clients, opens inside ten seconds of the send); human opens are `unique_opens` minus `machine_opens`. `machine_clicks` counts steps whose only clicks were automated (a security gateway walking the links); those are not part of `unique_clicks` or `total_clicks`, which only ever count a person's click. ## Get campaign daily stats diff --git a/docs/content/docs/api/reference/contacts.mdx b/docs/content/docs/api/reference/contacts.mdx index b0eeff2e..65548475 100644 --- a/docs/content/docs/api/reference/contacts.mdx +++ b/docs/content/docs/api/reference/contacts.mdx @@ -537,6 +537,8 @@ Auth: **Scope** `READ_CONTACTS` · **Org permission** `view_contacts` } ``` +`engagement.total_opened` and `engagement.last_opened_at` count opens by a person; opens a mail client or security gateway fetched automatically are left out, as they are in campaign analytics. A person's click counts as an open too; an automated click does not add to either field. + When the contact is suppressed, `suppression` is an object: `{ "id", "kind", "value", "reason", "source", "expires_at", "created_at" }`. `kind` is `email` when the contact's own address is on the list or `domain` when its whole domain is, `value` is the matching entry, and `source` is `bounce`, `complaint`, `unsubscribe`, `manual`, or `import`. `id` is the suppression entry, which `DELETE /suppressions/:id` lifts; see [deliverability and ops](/api/reference/deliverability-ops/#list-the-suppression-list). ## Update a contact @@ -738,7 +740,7 @@ Returns a `data` array and a `has_more` flag (not a cursor envelope). Paginate b `type` is one of `email_sent`, `email_opened`, `email_clicked`, `email_replied`, `email_bounced`, `reply_received`, `deliverability`, `suppressed`, `note`, `meeting_booked`, `meeting_rescheduled`, `meeting_canceled`, `contact_created`, `campaign_added`, `campaign_removed`, `category_added`, or `category_removed`. Fields not relevant to an event type are omitted. -`email_opened` and `email_clicked` carry `machine`: `true` when an automated fetcher did it rather than the person (a mail privacy proxy, a fetch inside ten seconds of the send, several links followed within seconds). Per-link `email_clicked` events also carry `machine_reason` (`prefetch`, `instant` or `burst`); opens carry only the flag. Automated clicks are on the feed for the record but never count the step as clicked. An `email_clicked` event carries `link` with the link's `id`, `url`, `label` (its anchor text), the `utm_*` parameters the URL carried, and the `user_agent`; every link in an email is tracked on its own, so each link clicked is its own event. Clicks recorded before per-link attribution have no `link`. +`email_opened` and `email_clicked` carry `machine`: `true` when an automated fetcher did it rather than the person (a mail privacy proxy, a fetch inside ten seconds of the send, several links followed within seconds). Per-link `email_clicked` events carry `machine_reason` (`prefetch`, `instant` or `burst`), and per-event `email_opened` rows carry it too (`prefetch` or `instant`); an open summarised from the lead alone carries only the flag. Both kinds carry an `origin` object when the event was logged: `client` when the user agent names a mail client or image proxy, `device_type`, `os`, `browser`, `browser_version`, and `country_code`, `region`, `city` when the consumer could resolve them. Opens appear once per event, so a contact who opened from two devices has two rows. Automated clicks are on the feed for the record but never count the step as clicked. An `email_clicked` event carries `link` with the link's `id`, `url`, `label` (its anchor text), the `utm_*` parameters the URL carried, and the `user_agent`; every link in an email is tracked on its own, so each link clicked is its own event. Clicks recorded before per-link attribution have no `link`. Lifecycle events carry the name of what changed as it was at the time (`campaign_name`, or `category_id` plus `category_title`), so a later rename or deletion does not rewrite history. A `contact_created` event carries `source` (`manual`, `campaign`, `import`, `sheet_sync`, `api`, `ai_assistant`, or `unknown` for contacts that predate attribution) and `source_detail` (the file, campaign, sheet or API key name). The same values are on the contact itself as `source`, `source_detail` and `first_seen_at`, and never change after creation. diff --git a/docs/content/docs/development/bare-metal.mdx b/docs/content/docs/development/bare-metal.mdx index c6dc37ff..f4bb4706 100644 --- a/docs/content/docs/development/bare-metal.mdx +++ b/docs/content/docs/development/bare-metal.mdx @@ -333,6 +333,9 @@ PHX_HOST=ws.example.com PORT=4000 CHECK_ORIGIN=true +# ── GeoIP (the variable is required by the backend; the file is optional) ── +GEODB_PATH=/var/lib/warmbly/GeoLite2-City.mmdb + # ── Platform email (resets, invitations, digests) ──────────────── MAIL_TRANSPORT=smtp SMTP_HOST=smtp.example.com @@ -357,7 +360,7 @@ A few lines differ from the compose defaults on purpose: - `API_HOST`, `TRACKING_HOST` and the realtime `PORT` sit on `127.0.0.1`, so only nginx reaches them. The realtime service binds all interfaces regardless, so keep `4000` closed at the firewall - `TRUSTED_PROXIES` and `TRACKING_TRUSTED_PROXIES` name the proxy, so rate limits and audit records see the visitor's address instead of `127.0.0.1` - `CHECK_ORIGIN=true` makes the websocket refuse browsers that are not on `PHX_HOST`'s origin list -- `GEODB_PATH` is left unset. It only adds a city to sessions and audit rows; set it to a MaxMind `GeoLite2-City.mmdb` if you have one +- `GEODB_PATH` points at a file that need not exist. The backend refuses to start without the variable, but a missing database only means sessions, audit rows and email opens and clicks carry no city; drop a MaxMind `GeoLite2-City.mmdb` at that path if you have one Every other variable, with its default, is in the [configuration reference](/development/configuration/), and [`deploy/config/env.example`](https://github.com/warmbly/warmbly/blob/main/deploy/config/env.example) is the annotated template this file was cut down from. Mail relay, OAuth clients, single sign-on and the AI provider are configured exactly as in the [self-hosting guide](/development/deployment-guide/#platform-email); only the way values reach the process differs. diff --git a/docs/content/docs/development/configuration.mdx b/docs/content/docs/development/configuration.mdx index 268d3a27..aa5de654 100644 --- a/docs/content/docs/development/configuration.mdx +++ b/docs/content/docs/development/configuration.mdx @@ -305,7 +305,7 @@ Redis holds rate limit counters, the organization key cache, the realtime bridge |---|---|---|---| | `GEODB_PATH` | Path to a GeoLite2 City database | none, and the backend refuses to start without the variable | yes | -The variable must be set on the backend in every environment. The file itself is optional: a missing file at that path means sessions and audit rows are recorded without a city, and nothing else changes. +The variable must be set on the backend in every environment. The file itself is optional: a missing file at that path means sessions and audit rows are recorded without a city, and nothing else changes. The consumer reads the same variable, optionally, to put a country and city on each email open and click; without it those records carry client and device only. ## Workers @@ -420,6 +420,7 @@ The Rust open and click service. It reads its own environment, so these have to | `TRACKING_RATE_LIMIT_PER_MIN` | Counted pixel and click requests per source per minute. Over budget, pixels are still served but not counted, and click redirects get `429` | `300` | | `TRACKING_PAGEHIT_RATE_LIMIT_PER_MIN` | Website page views accepted per source per minute, on top of the shared budget above. Over budget, the snippet gets `429` | `60` | | `TRACKING_TRUSTED_PROXIES` | CIDRs the tracking service accepts a forwarded client address from. Empty trusts nothing and uses the socket peer, which is correct for a directly exposed service; set it behind a reverse proxy or the per-source rate limits and the location stored with page views are caller-controlled. Same convention as the backend's `TRUSTED_PROXIES` | empty | +| `TRACKING_IP_HASH_KEY` | Secret the source-address token in tracking events is keyed with. The token names one source for deduplication, rate limits and the click burst rule; keyed, it cannot be turned back into the address by enumeration | `INTERNAL_API_TOKEN` | | `TRACKING_CLIENT_IP_HEADER` | The one header a trusted proxy sets with the client address. No other header is read, so a caller cannot smuggle an address past a generic proxy in `CF-Connecting-IP`. For `x-forwarded-for` the proxy-appended last entry is used; set `cf-connecting-ip` behind Cloudflare | `x-forwarded-for` | | `EVENTBUS_PROVIDER` | `nats` or `kafka`. Kafka needs an image built with `CARGO_FEATURES=kafka` | `nats` | | `NATS_URL`, `NATS_SUBJECT_PREFIX` | JetStream address and subject prefix. The publish subject is `.` | `nats://localhost:4222`, `warmbly` | diff --git a/docs/content/docs/development/deployment-guide.mdx b/docs/content/docs/development/deployment-guide.mdx index 86a3cef6..7609afbe 100644 --- a/docs/content/docs/development/deployment-guide.mdx +++ b/docs/content/docs/development/deployment-guide.mdx @@ -226,7 +226,7 @@ These are the ones that break things quietly when they drift, because each servi | realtime | `JWT_SECRET`, `SECRET_KEY_BASE`, `DATABASE_URL`, `REDIS_URL`, `PHX_HOST` | | web / admin | Only `WARMBLY_*` URLs, read at container start and written into `/config.js`. The same image runs anywhere | -`APP_ENV` accepts `dev` or `prod`. No other value turns on production behavior. `prod` needs no cloud account: `SENTRY_DSN` stays optional in every environment, and `GEODB_PATH` must be *set* on the backend everywhere while the file it points at is optional. +`APP_ENV` accepts `dev` or `prod`. No other value turns on production behavior. `prod` needs no cloud account: `SENTRY_DSN` stays optional in every environment, and `GEODB_PATH` must be *set* on the backend everywhere while the file it points at is optional (the consumer reads it too, optionally, for the location on opens and clicks). The complete list of variables, with defaults and whether a change needs a restart, is the [configuration reference](/development/configuration/). diff --git a/docs/content/docs/development/instance-health.mdx b/docs/content/docs/development/instance-health.mdx index 36c97212..24355e84 100644 --- a/docs/content/docs/development/instance-health.mdx +++ b/docs/content/docs/development/instance-health.mdx @@ -252,7 +252,7 @@ A reachable realtime service that still leaves the dashboard dead has a differen | Service | Needs | |---|---| | backend | The five secrets, `PRIMARY_DB`, `REDIS`, the provider switches, the public URLs, `EMAIL_ADDRESS`, `EMAIL_NAME` and `GEODB_PATH` | -| consumer | The same shared block. It writes to Postgres, so it needs `PRIMARY_DB` and both encryption keys, and it needs the mail identity or it silently sends nothing | +| consumer | The same shared block. It writes to Postgres, so it needs `PRIMARY_DB` and both encryption keys, and it needs the mail identity or it silently sends nothing. `GEODB_PATH` is optional and only adds a location to opens and clicks | | worker | No database. The event bus, `REDIS`, both encryption keys, `ENCRYPTED_KEYS_BACKEND_URL` plus the worker token, and the `BOX_*` OAuth clients | | tracking | The event bus, plus `BACKEND_INTERNAL_URL` and `INTERNAL_API_TOKEN`. It exits at boot without either of those two | | realtime | `JWT_SECRET` equal to the backend's `AUTH_SECRET`, plus `SECRET_KEY_BASE` and `DATABASE_URL`. It refuses to boot without all three. `REDIS_URL`, `PHX_HOST` and the connection limits have defaults | diff --git a/docs/content/docs/guides/analytics.mdx b/docs/content/docs/guides/analytics.mdx index cd0130b6..59a7a95f 100644 --- a/docs/content/docs/guides/analytics.mdx +++ b/docs/content/docs/guides/analytics.mdx @@ -22,6 +22,7 @@ Four rules govern the counts: - **Replies are human replies.** Out-of-office and autoresponders never count, never stamp the contact as replied, and never trip stop-on-reply. - **Bots are filtered.** Crawlers, CLI agents, prefetches, chat link previews, and security gateways that announce themselves are served normally but never counted. The ones that do not announce themselves are caught by what they do: an open or click inside ten seconds of the send (nobody reads that fast; the clock starts when the send is handed to the worker, before the mail has even been delivered) and clicks on two or more links of one email from the same source within five seconds (a scanner walking the message). Otherwise one corporate scanner would "click" every link seconds after delivery. - **Auto-opens are labeled, not hidden.** Privacy proxies like Apple Mail Privacy Protection, and instant opens, still count (they confirm delivery) but are tagged and shown separately (`12 auto`). A later real open upgrades them to human. **Auto-opens never trigger opened-based branches or automations.** +- **A click is an open.** Opens need the mail client to load images, and many clients block them. A click by the person proves the email was read, so it counts as an open as well, and a lead can never read "clicked, not opened". - **Auto-clicks are kept but never counted.** A click classified as automated is logged on the contact's activity with the link it hit and the rule that caught it, and the campaign overview shows how many steps had only automated clicks (`3 auto`). It never makes the step clicked, never fires a clicked branch or automation, and never sends a webhook. A person clicking the same link later counts normally. A burst is only recognisable from its second click, so a click that looks human waits for the burst window plus one second (six seconds) before it fires anything (clicked branches, automations, webhooks, the live feed): if a second link follows inside that window, both clicks are relabeled, the stamp is withdrawn and nothing fires. ## Workspace dashboard @@ -43,6 +44,14 @@ Each campaign reports its own totals: contacts, sent, pending, unique opens and **Per-step stats** break sent, opens, clicks, replies, and bounces down by sequence step, which is how you find the touch pulling replies and the follow-ups that are dead weight. Daily stats drive the trend chart, and hourly stats show when sends land and get engagement. Campaigns can also be compared side by side on the same metrics. +**Who engaged, and from where** lists the countries, mail clients and devices people opened and clicked from, counting each contact once per bucket. Only human events feed it, so a security scanner's data centre never tops the country list. + +### What is recorded per open and click + +Every open and every click is kept as its own record, repeats included, with what the request said about itself: the mail client or image proxy when the user agent names one (Gmail, Apple Mail, Outlook), otherwise the browser, operating system and device type; the country, region and city resolved from the source network; for clicks the exact link; and whether it was the person or an automated fetch, with the rule that caught it. The address itself never leaves the tracking service: it publishes the network the address belongs to (the last IPv4 octet zeroed, an IPv6 address cut to its first 48 bits), which is enough for a city-level lookup and cannot single out a host, plus a keyed token of the full address used to tell one source from another, which cannot be turned back into the address. The consumer resolves the network to a location and stores neither. Location needs the GeoLite2 database on the consumer (`GEODB_PATH`); without it the country, region and city stay empty and everything else is still recorded. These records are kept for a year, then pruned; the counts and the first open and click per step live on the lead itself and are not affected. + +These records back the engagement rows on a contact's [activity timeline](/guides/contacts-crm/#activity-timeline) and the per-campaign breakdown above. + ## Per-mailbox | Signal | What it tells you | diff --git a/docs/content/docs/guides/campaigns.mdx b/docs/content/docs/guides/campaigns.mdx index ffecbac4..4d134812 100644 --- a/docs/content/docs/guides/campaigns.mdx +++ b/docs/content/docs/guides/campaigns.mdx @@ -95,7 +95,9 @@ A lead is **Processing** only while steps remain, so a finished campaign reads a ### Who opened, clicked and replied -Next to each lead's status, the Leads list shows three engagement columns: **Opened**, **Clicked** and **Replied**, each with the number of emails in the sequence the person engaged with. A dash means the lead was emailed and has not engaged; the cell is blank for a lead not emailed yet. An open counts only when a person opened the email. Mail clients that fetch every image automatically (Apple Mail Privacy Protection, for example) show as **auto** instead, the same opens the campaign overview reports as automatic, so they never pass for engagement. Clicks are held to the same standard: a link followed within ten seconds of the send, or several links of one email followed within a few seconds of each other, is a security gateway scanning the message, not the recipient. Those clicks are kept on the contact's activity marked **auto**, counted apart on the campaign overview, and never make a lead **Clicked**, never fire a clicked branch or automation, and never send a webhook. See [Link tracking and UTM parameters](#link-tracking-and-utm-parameters). +Next to each lead's status, the Leads list shows three engagement columns: **Opened**, **Clicked** and **Replied**, each with the number of emails in the sequence the person engaged with. A dash means the lead was emailed and has not engaged; the cell is blank for a lead not emailed yet. An open counts when a person opened the email, or clicked a link in it. Mail clients that fetch every image automatically (Apple Mail Privacy Protection, for example) show as **auto** instead, the same opens the campaign overview reports as automatic, so they never pass for engagement. Clicks are held to the same standard: a link followed within ten seconds of the send, or several links of one email followed within a few seconds of each other, is a security gateway scanning the message, not the recipient. Those clicks are kept on the contact's activity marked **auto**, counted apart on the campaign overview, and never make a lead **Clicked**, never fire a clicked branch or automation, and never send a webhook. See [Link tracking and UTM parameters](#link-tracking-and-utm-parameters). + +Opens depend on the recipient's mail client loading images. Outlook, many corporate setups and some privacy-minded clients block them, so a person can read an email and still show no open; the info icon on the column says as much. A click is stronger evidence, so a click by the person always counts as an open too. Open the contact's Activity tab to see each open and click with the mail client and location it came from. The chips above the list are filters. Click a status chip (**Processing**, **Done**, **Replied**, **Queued**, **Bounced**, **Unsub**) or an engagement chip (**Opened**, **Not opened**, **Clicked**, **Not clicked**, **Replied**, **Not replied**) to show only those leads; click it again to clear. One status and one engagement chip can be active at once and both must match. The numbers on the chips are campaign-wide totals, and filtering happens on the server, so a scope shows every matching lead however long the list is. **Not opened**, **Not clicked** and **Not replied** only cover leads that have been sent at least one email: a queued lead has not had the chance. The same filters live in the **Filters** sheet under **Lead status** and **Engagement**. diff --git a/docs/content/docs/guides/contacts-crm.mdx b/docs/content/docs/guides/contacts-crm.mdx index d42bafa7..ef6dcd66 100644 --- a/docs/content/docs/guides/contacts-crm.mdx +++ b/docs/content/docs/guides/contacts-crm.mdx @@ -109,7 +109,7 @@ Only new contacts get a source. An import or API call that matches an existing c The **Activity** tab of a contact is one feed, newest first, of everything Warmbly knows about them: every campaign email sent, opened, clicked, replied to or bounced (with the campaign, step, subject and sending mailbox), replies with their classified intent, deliverability and suppression events, notes, meetings, and the contact's lifecycle: when it was created and how, and each time it joined or left a campaign or a category. -A click names the link: the row reads **Clicked Pricing** and, expanded, shows the link's text, its full URL, the UTM source, medium, campaign and content it carried, and the browser. Every link in an email is tracked on its own, so two links clicked are two rows. Opens and clicks that came from a machine rather than the person (a mail privacy proxy, a security gateway that follows every link at delivery) carry an **auto** badge, and the expanded row says which rule caught them; see [Link tracking and UTM parameters](/guides/campaigns/#link-tracking-and-utm-parameters). +Opens appear once per event, not once per email: a second open from another device is its own row. Opens and clicks show the mail client or browser they came from and the city and country when known (see [what is recorded per open and click](/guides/analytics/#what-is-recorded-per-open-and-click)); expanded, the operating system, device and full location. A click names the link: the row reads **Clicked Pricing** and, expanded, shows the link's text, its full URL, the UTM source, medium, campaign and content it carried, and the browser. Every link in an email is tracked on its own, so two links clicked are two rows. Opens and clicks that came from a machine rather than the person (a mail privacy proxy, a security gateway that follows every link at delivery) carry an **auto** badge, and the expanded row says which rule caught them; see [Link tracking and UTM parameters](/guides/campaigns/#link-tracking-and-utm-parameters). Filter chips narrow the feed (**Emails**, **Replies**, **Deliv.**, **Notes**, **Meetings**, **Campaigns**, **Lifecycle**), the search box matches subjects, campaigns, steps, mailboxes, categories, reasons and, for clicks, the link's text, URL and UTM values, and the date picker bounds it. Each row stays to one line until you click it; expanded, it shows every detail the event carries. The feed updates live as teammates and the schedulers write to it. diff --git a/docs/content/docs/guides/workspace-export-import.mdx b/docs/content/docs/guides/workspace-export-import.mdx index 6d2cc642..1869eb7e 100644 --- a/docs/content/docs/guides/workspace-export-import.mdx +++ b/docs/content/docs/guides/workspace-export-import.mdx @@ -15,7 +15,7 @@ The data is split into groups. Every export includes **Workspace**; the rest are |-------|----------| | Workspace | The organization, members, roles, teams, mailboxes, API keys, webhooks, and settings, including the website tracking site key. Always included | | Contacts | Contacts, categories, segments with their manual overrides, forms with their images, submissions, personalized link tickets and funnel events, notes, activities, and the suppression list | -| Campaigns | Campaigns, sequences, senders, linked segments, attachments, per-campaign settings, and each lead's step progress with its per-link clicks | +| Campaigns | Campaigns, sequences, senders, linked segments, attachments, per-campaign settings, and each lead's step progress with its per-link clicks and per-event opens | | CRM | Pipelines, deals, tasks, and meeting bookings | | Automations | Automations, connected integrations, and lead sync sources | | Assistant | Assistant sessions and messages, skills, MCP servers, and AI settings | diff --git a/internal/app/analytics/service.go b/internal/app/analytics/service.go index 41fa49b4..14aa44e2 100644 --- a/internal/app/analytics/service.go +++ b/internal/app/analytics/service.go @@ -132,12 +132,19 @@ func (s *analyticsService) GetCampaignAnalytics(ctx context.Context, userID, cam return nil, xerr } + // Where and on what people engaged; best-effort, the totals stand alone. + engagement, xerr := s.analyticsRepo.GetCampaignEngagementBreakdown(ctx, campaignID, 8) + if xerr != nil { + engagement = nil + } + return &models.CampaignAnalytics{ CampaignID: campaignID, Name: campaign.Name, Status: campaign.Status, Summary: *summary, Sequences: sequences, + Engagement: engagement, }, nil } diff --git a/internal/app/consumer/event_tracking.go b/internal/app/consumer/event_tracking.go index bcfa388b..a4c47814 100644 --- a/internal/app/consumer/event_tracking.go +++ b/internal/app/consumer/event_tracking.go @@ -4,9 +4,12 @@ import ( "context" "crypto/sha256" "encoding/hex" + "net/netip" + "strings" "time" "github.com/google/uuid" + "github.com/mileusna/useragent" "github.com/rs/zerolog/log" "github.com/warmbly/warmbly/internal/app/advanced" "github.com/warmbly/warmbly/internal/config" @@ -15,6 +18,7 @@ import ( "github.com/warmbly/warmbly/internal/infrastructure/eventbus" "github.com/warmbly/warmbly/internal/infrastructure/pubsub" "github.com/warmbly/warmbly/internal/models" + "github.com/warmbly/warmbly/internal/pkg/geo" "github.com/warmbly/warmbly/internal/repository" ) @@ -41,8 +45,12 @@ type TrackingConsumer struct { // ProcessIncomingReply). Best-effort and nil-safe: when unset, opens/clicks // are still recorded and routed at the next step boundary by the scheduler. advancedService advanced.Service - topic string - group string + // opens is the per-event open log; geo resolves a source network to a + // location for opens and clicks. Both optional. + opens repository.EmailOpenRepository + geo *geo.Client + topic string + group string } // NewTrackingConsumer wires the tracking consumer to the shared event bus. @@ -60,6 +68,8 @@ func NewTrackingConsumer( linkClicks repository.LinkClickRepository, advancedService advanced.Service, evidence advanced.EvidenceRecorder, + opens repository.EmailOpenRepository, + geoClient *geo.Client, ) (*TrackingConsumer, error) { return &TrackingConsumer{ bus: bus, @@ -79,16 +89,94 @@ func NewTrackingConsumer( }, advancedService: advancedService, evidence: evidence, + opens: opens, + geo: geoClient, topic: topic, group: group, }, nil } // Start subscribes to the tracking topic and blocks until ctx is cancelled. +// It also runs the daily prune of the open and click logs. func (tc *TrackingConsumer) Start(ctx context.Context) error { + if tc.opens != nil || tc.linkClicks != nil { + go tc.pruneEngagementLogs(ctx) + } + if tc.linkClicks != nil { + go tc.sweepPendingClicks(ctx) + } return tc.bus.Subscribe(ctx, []string{tc.topic}, tc.group, tc.receive) } +// sweepPendingClicks fires the effects of human clicks whose timer never +// ran or never finished: the consumer restarted inside the burst window, a +// claim failed, or an attempt died mid-way and its lease expired. Runs at +// start and every minute until ctx ends. A click the timer completed is not +// offered, and one under a live lease is not offered twice. +func (tc *TrackingConsumer) sweepPendingClicks(ctx context.Context) { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for { + before := time.Now().Add(-time.Duration(config.TrackingClickBurstSeconds+1) * time.Second) + pending, err := tc.linkClicks.ListPendingAnnouncements(ctx, before, 200) + if err != nil { + log.Warn().Err(err).Msg("could not list pending click announcements") + } + for i := range pending { + c := &pending[i] + task := &repository.CampaignTask{TaskID: c.TaskID, CampaignID: &c.CampaignID, ContactID: &c.ContactID, SequenceID: &c.SequenceID} + destination := c.Destination + event := events.TrackingEvent{ + EventType: events.EventTypeEmailClicked, + TaskID: c.TaskID.String(), + OriginalURL: &destination, + Timestamp: c.ClickedAt.Format(time.RFC3339Nano), + } + if c.TrackedLinkID != nil { + id := c.TrackedLinkID.String() + event.LinkID = &id + } + tc.finishHumanClick(task, event, c.ID, c.Label, c.Origin) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +// pruneEngagementLogs deletes opens and clicks older than the retention +// window, at start and then daily. The progress-row summary stays, so +// nothing a count, filter or branch reads is affected. +func (tc *TrackingConsumer) pruneEngagementLogs(ctx context.Context) { + ticker := time.NewTicker(24 * time.Hour) + defer ticker.Stop() + for { + pctx, cancel := context.WithTimeout(ctx, 5*time.Minute) + if tc.opens != nil { + if n, err := tc.opens.Cleanup(pctx, config.EngagementEventRetentionDays); err != nil { + log.Warn().Err(err).Msg("open log prune failed") + } else if n > 0 { + log.Info().Int64("deleted", n).Msg("open log pruned") + } + } + if tc.linkClicks != nil { + if n, err := tc.linkClicks.Cleanup(pctx, config.EngagementEventRetentionDays); err != nil { + log.Warn().Err(err).Msg("click log prune failed") + } else if n > 0 { + log.Info().Int64("deleted", n).Msg("click log pruned") + } + } + cancel() + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + // Close is a no-op: the event bus lifecycle is owned by the consumer main, // which subscribes both worker-events and tracking on the same bus. func (tc *TrackingConsumer) Close() {} @@ -162,7 +250,7 @@ func (tc *TrackingConsumer) HandleTrackingEvent(ctx context.Context, event *even var reason string switch event.EventType { case events.EventTypeEmailOpened: - machine = isMachineOpen(event.UserAgent) || isInstant(sentAt, at) + machine, reason = classifyOpen(event.UserAgent, sentAt, at) case events.EventTypeEmailClicked: machine, reason = classifyClick(event.UserAgent, sentAt, at) default: @@ -170,6 +258,10 @@ func (tc *TrackingConsumer) HandleTrackingEvent(ctx context.Context, event *even return nil } + // What the request said about where it came from, for the logs and the + // live feed. The source network is resolved here and goes no further. + origin := tc.originOf(event) + // Check for duplicate at consumer level (belt and suspenders with Rust service) if tc.dedupeRepo != nil { processed, err := tc.dedupeRepo.IsProcessed(ctx, taskID, event.EventType, urlHash) @@ -181,6 +273,11 @@ func (tc *TrackingConsumer) HandleTrackingEvent(ctx context.Context, event *even // label (a gateway scanned at delivery; the person acted later). // Quiet write only: the event was already counted once, so no // automations and no re-publish. + if event.EventType == events.EventTypeEmailOpened { + // Every open is logged, repeats and machines included: a + // second open from another device is worth seeing. + tc.logOpen(ctx, campaignTask, event, at, machine, reason, origin) + } if machine { return nil } @@ -188,7 +285,7 @@ func (tc *TrackingConsumer) HandleTrackingEvent(ctx context.Context, event *even case events.EventTypeEmailOpened: _ = tc.campaignProgressRepo.RecordEmailOpened(ctx, campaignID, contactID, sequenceID, false) case events.EventTypeEmailClicked: - tc.upgradeClick(ctx, campaignTask, event, at) + tc.upgradeClick(ctx, campaignTask, event, at, origin) } return nil } @@ -205,6 +302,7 @@ func (tc *TrackingConsumer) HandleTrackingEvent(ctx context.Context, event *even switch event.EventType { case events.EventTypeEmailOpened: err = tc.campaignProgressRepo.RecordEmailOpened(ctx, campaignID, contactID, sequenceID, machine) + tc.logOpen(ctx, campaignTask, event, at, machine, reason, origin) if !machine { instantKind = "open" // A human open proves the mailbox is live; a prefetch proves @@ -215,7 +313,7 @@ func (tc *TrackingConsumer) HandleTrackingEvent(ctx context.Context, event *even } case events.EventTypeEmailClicked: var click *repository.LinkClick - machine, reason, click, err = tc.recordClick(ctx, campaignTask, event, at, machine, reason) + machine, reason, click, err = tc.recordClick(ctx, campaignTask, event, at, machine, reason, origin) if click != nil { linkLabel = click.Label } @@ -226,7 +324,7 @@ func (tc *TrackingConsumer) HandleTrackingEvent(ctx context.Context, event *even if err == nil && click != nil && tc.afterBurstWindow != nil { deferred = true task, ev, clickID, label := campaignTask, *event, click.ID, linkLabel - tc.afterBurstWindow(func() { tc.finishHumanClick(task, ev, clickID, label) }) + tc.afterBurstWindow(func() { tc.finishHumanClick(task, ev, clickID, label, origin) }) } else if err == nil { instantKind = "click" if tc.evidence != nil { @@ -264,7 +362,7 @@ func (tc *TrackingConsumer) HandleTrackingEvent(ctx context.Context, event *even // Publish to Pub/Sub for realtime updates (a deferred human click // publishes once its verdict is final) if !deferred { - tc.publishTrackingEvent(ctx, campaignTask, *event, machine, linkLabel) + tc.publishTrackingEvent(ctx, campaignTask, *event, machine, linkLabel, origin) } return nil @@ -272,21 +370,28 @@ func (tc *TrackingConsumer) HandleTrackingEvent(ctx context.Context, event *even // finishHumanClick runs the effects of a click that looked human when it // landed, once the burst window has passed: if a burst relabelled it in the -// meantime it is announced as automated and nothing else fires. The stamp, -// the click log and the dedupe mark were written up front, so a consumer -// restart inside the window loses only these effects, and the scheduler -// still routes the clicked branch at the next step boundary. The verdict is -// re-read with retries and, when it cannot be read at all, nothing fires: -// an automation for a scanner's click is worse than a missed one, and the -// step boundary still routes on the stored stamp. -func (tc *TrackingConsumer) finishHumanClick(task *repository.CampaignTask, event events.TrackingEvent, clickID uuid.UUID, label string) { +// meantime it is announced as automated and nothing else fires. The click +// row carries the pending flag, written before the event was marked +// processed, so a consumer restart inside the window hands the click to +// sweepPendingClicks instead of losing it, and a redelivery cannot fire it +// twice. When the claim cannot be made at all, nothing fires here and the +// sweep retries: an automation for a scanner's click is worse than a late +// one, and the step boundary still routes on the stored stamp. +func (tc *TrackingConsumer) finishHumanClick(task *repository.CampaignTask, event events.TrackingEvent, clickID uuid.UUID, label string, origin models.EngagementOrigin) { ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) defer cancel() - var machine bool + // The claim is the verdict and the lease in one write: a burst that + // relabelled the click in the meantime shows up as machine, and a click + // already announced, or under another attempt's live lease, is not + // claimed. The row is completed only after the effects ran, so a crash + // in between is retried by the sweep once the lease expires (at-least- + // once: the instant actions claim their own once-only fire, the rest + // may repeat after a crash). + var claimed, machine bool var err error for attempt := 0; attempt < 5; attempt++ { - if machine, err = tc.linkClicks.IsMachine(ctx, clickID); err == nil { + if claimed, machine, err = tc.linkClicks.ClaimAnnounce(ctx, clickID); err == nil { break } select { @@ -295,20 +400,26 @@ func (tc *TrackingConsumer) finishHumanClick(task *repository.CampaignTask, even } } if err != nil { - log.Error().Err(err).Str("click_id", clickID.String()).Msg("could not re-read click classification; click effects skipped") + log.Error().Err(err).Str("click_id", clickID.String()).Msg("could not claim click announcement; left for the sweep") + return + } + if !claimed { return } if machine { - tc.publishTrackingEvent(ctx, task, event, true, label) - return + tc.publishTrackingEvent(ctx, task, event, true, label, origin) + } else { + if tc.evidence != nil { + tc.evidence.RecordEvidence(ctx, *task.ContactID, "clicked", task.SequenceID.String(), "") + } + if tc.advancedService != nil { + tc.advancedService.FireInstantActions(ctx, *task.CampaignID, *task.ContactID, *task.SequenceID, "click") + } + tc.publishTrackingEvent(ctx, task, event, false, label, origin) } - if tc.evidence != nil { - tc.evidence.RecordEvidence(ctx, *task.ContactID, "clicked", task.SequenceID.String(), "") + if err := tc.linkClicks.CompleteAnnounce(ctx, clickID); err != nil { + log.Warn().Err(err).Str("click_id", clickID.String()).Msg("could not mark click announcement complete; the sweep may repeat it after the lease") } - if tc.advancedService != nil { - tc.advancedService.FireInstantActions(ctx, *task.CampaignID, *task.ContactID, *task.SequenceID, "click") - } - tc.publishTrackingEvent(ctx, task, event, false, label) } // resolveLink names the clicked link: the minted ticket when the event @@ -342,7 +453,7 @@ func (tc *TrackingConsumer) resolveLink(ctx context.Context, event *events.Track // that leaves the step with no human click, the clicked stamp the first // click already wrote is walked back. Returns the final classification and // the logged row (nil when nothing could be logged). -func (tc *TrackingConsumer) recordClick(ctx context.Context, task *repository.CampaignTask, event *events.TrackingEvent, at time.Time, machine bool, reason string) (bool, string, *repository.LinkClick, error) { +func (tc *TrackingConsumer) recordClick(ctx context.Context, task *repository.CampaignTask, event *events.TrackingEvent, at time.Time, machine bool, reason string, origin models.EngagementOrigin) (bool, string, *repository.LinkClick, error) { if tc.linkClicks == nil { return machine, reason, nil, nil } @@ -383,6 +494,11 @@ func (tc *TrackingConsumer) recordClick(ctx context.Context, task *repository.Ca Machine: machine, MachineReason: reason, ClickedAt: at, + Origin: origin, + // A person's click waits out the burst window; the flag is the + // durable record of that, written before the event is marked + // processed, so a restart or a redelivery fires it exactly once. + AnnouncePending: !machine && tc.afterBurstWindow != nil, } if err := tc.linkClicks.Insert(ctx, click); err != nil { return machine, reason, click, err @@ -402,7 +518,7 @@ func (tc *TrackingConsumer) recordClick(ctx context.Context, task *repository.Ca // upgradeClick handles a human click on a link this email was already // credited for: the step is stamped clicked if only machines had clicked so // far, and the click is logged once so the timeline shows the person's. -func (tc *TrackingConsumer) upgradeClick(ctx context.Context, task *repository.CampaignTask, event *events.TrackingEvent, at time.Time) { +func (tc *TrackingConsumer) upgradeClick(ctx context.Context, task *repository.CampaignTask, event *events.TrackingEvent, at time.Time, origin models.EngagementOrigin) { _ = tc.campaignProgressRepo.RecordEmailClicked(ctx, *task.CampaignID, *task.ContactID, *task.SequenceID) if tc.linkClicks == nil { return @@ -432,12 +548,73 @@ func (tc *TrackingConsumer) upgradeClick(ctx context.Context, task *repository.C UserAgent: userAgent, IPHash: ipHash, ClickedAt: at, + Origin: origin, }) } +// logOpen writes one row to the open log for this event, whatever it was +// classified as; the label travels with it. +func (tc *TrackingConsumer) logOpen(ctx context.Context, task *repository.CampaignTask, event *events.TrackingEvent, at time.Time, machine bool, reason string, origin models.EngagementOrigin) { + if tc.opens == nil { + return + } + open := &repository.EmailOpen{ + TaskID: task.TaskID, + CampaignID: *task.CampaignID, + ContactID: *task.ContactID, + SequenceID: *task.SequenceID, + OpenedAt: at, + Machine: machine, + MachineReason: reason, + Origin: origin, + } + if event.UserAgent != nil { + open.UserAgent = clipString(*event.UserAgent, 512) + } + if event.IPHash != nil { + open.IPHash = *event.IPHash + } + if err := tc.opens.Insert(ctx, open); err != nil { + log.Warn().Err(err).Str("task_id", task.TaskID.String()).Msg("failed to log open") + } +} + +// originOf reads what the event says about its source: the user agent parsed +// to client, browser and device, and the source network resolved to a +// location. The network is used here and dropped. +func (tc *TrackingConsumer) originOf(event *events.TrackingEvent) models.EngagementOrigin { + var o models.EngagementOrigin + if event.UserAgent != nil && strings.TrimSpace(*event.UserAgent) != "" { + ua := useragent.Parse(*event.UserAgent) + o.OS, o.Browser, o.BrowserVersion = ua.OS, ua.Name, ua.Version + o.DeviceType = deviceType(ua) + o.Client = clientName(*event.UserAgent) + } + if event.ClientIP != nil && tc.geo != nil { + if addr, err := netip.ParseAddr(strings.TrimSpace(*event.ClientIP)); err == nil && !addr.IsPrivate() && !addr.IsLoopback() { + if info, err := tc.geo.Lookup(addr); err == nil && info != nil { + o.CountryCode = info.CountryCode + o.Region = info.Region + if info.City != "Unknown" { + o.City = info.City + } + } + } + } + return o +} + +func clipString(s string, n int) string { + s = strings.TrimSpace(s) + if len(s) > n { + return s[:n] + } + return s +} + // publishTrackingEvent publishes the tracking event to Pub/Sub for realtime UI // updates AND fans an opt-in firehose webhook (campaign.email_opened/clicked). -func (tc *TrackingConsumer) publishTrackingEvent(ctx context.Context, task *repository.CampaignTask, event events.TrackingEvent, machine bool, linkLabel string) { +func (tc *TrackingConsumer) publishTrackingEvent(ctx context.Context, task *repository.CampaignTask, event events.TrackingEvent, machine bool, linkLabel string, origin models.EngagementOrigin) { // Get campaign to find user ID + org campaign, err := tc.campaignRepo.GetByID(ctx, *task.CampaignID) if err != nil || campaign == nil { @@ -513,6 +690,11 @@ func (tc *TrackingConsumer) publishTrackingEvent(ctx context.Context, task *repo ContactEmail: contactEmail, SequenceID: task.SequenceID.String(), Machine: machine, + OccurredAt: eventTime(event.Timestamp), + Client: origin.Client, + DeviceType: origin.DeviceType, + CountryCode: origin.CountryCode, + City: origin.City, } if event.EventType == events.EventTypeEmailClicked && event.OriginalURL != nil { diff --git a/internal/app/consumer/open_class.go b/internal/app/consumer/open_class.go index 5a3ae783..e451aeea 100644 --- a/internal/app/consumer/open_class.go +++ b/internal/app/consumer/open_class.go @@ -4,6 +4,7 @@ import ( "strings" "time" + "github.com/mileusna/useragent" "github.com/warmbly/warmbly/internal/config" "github.com/warmbly/warmbly/internal/repository" ) @@ -69,3 +70,59 @@ func eventTime(stamp string) time.Time { } return time.Now() } + +// classifyOpen applies the per-event open rules and names the one that +// caught it: prefetch for a mail proxy or a fetch with no browser, instant +// for a fetch inside the machine window after dispatch. An empty reason is +// a person. +func classifyOpen(userAgent *string, sentAt *time.Time, at time.Time) (bool, string) { + if isMachineOpen(userAgent) { + return true, repository.EmailOpenReasonPrefetch + } + if isInstant(sentAt, at) { + return true, repository.EmailOpenReasonInstant + } + return false, "" +} + +// clientName names the mail client or image proxy behind a user agent when +// it says so; empty for a plain browser, which the parsed fields describe. +func clientName(userAgent string) string { + ua := strings.ToLower(strings.TrimSpace(userAgent)) + switch { + case ua == "": + return "" + case strings.Contains(ua, "googleimageproxy"): + return "Gmail" + case strings.Contains(ua, "yahoomailproxy"), strings.Contains(ua, "yahoo mail"): + return "Yahoo Mail" + case strings.Contains(ua, "outlook"), strings.Contains(ua, "microsoft office"): + return "Outlook" + case strings.Contains(ua, "thunderbird"): + return "Thunderbird" + case strings.Contains(ua, "superhuman"): + return "Superhuman" + case strings.Contains(ua, "protonmail"), strings.Contains(ua, "proton mail"): + return "Proton Mail" + case strings.Contains(ua, "hey.com"): + return "HEY" + case strings.HasSuffix(ua, "(khtml, like gecko)"): + // Apple Mail Privacy Protection's prefetch fingerprint. + return "Apple Mail" + } + return "" +} + +// deviceType folds the parser's flags into desktop, mobile, tablet or unknown. +func deviceType(ua useragent.UserAgent) string { + switch { + case ua.Tablet: + return "tablet" + case ua.Mobile: + return "mobile" + case ua.Desktop: + return "desktop" + default: + return "unknown" + } +} diff --git a/internal/app/orgtransfer/spec.go b/internal/app/orgtransfer/spec.go index f84fc0bf..aa4892cb 100644 --- a/internal/app/orgtransfer/spec.go +++ b/internal/app/orgtransfer/spec.go @@ -661,6 +661,13 @@ var Tables = []Table{ Name: "email_link_clicks", Group: models.OrgDataGroupCampaigns, Scope: `campaign_id IN ` + orgCampaigns, }, + { + // The per-event open log beside the click log: same keys, same + // scope, no ticket reference. task_id is an opaque id from the source + // instance, used only to group a step's opens. + Name: "email_opens", Group: models.OrgDataGroupCampaigns, + Scope: `campaign_id IN ` + orgCampaigns, + }, // ---------- delivery events ---------- { diff --git a/internal/config/constants.go b/internal/config/constants.go index 5f1d828a..5f8dc051 100644 --- a/internal/config/constants.go +++ b/internal/config/constants.go @@ -145,6 +145,11 @@ const ( // a scanner walking the message. A person follows one link at a time. TrackingClickBurstSeconds = 5 + // EngagementEventRetentionDays is how long the per-event open and click + // logs (client, device, location) are kept. The summary on the progress + // row outlives them, so counts and routing never change. + EngagementEventRetentionDays = 365 + // CampaignSendStampAttempts is how many times the control plane retries the // sent_at stamp after a send is already on the bus. The reservation is what // keeps the step from being re-sent, so a lost stamp is a pacing problem, diff --git a/internal/events/schemas.go b/internal/events/schemas.go index 12a8ca07..728e6a10 100644 --- a/internal/events/schemas.go +++ b/internal/events/schemas.go @@ -57,4 +57,10 @@ type TrackingEvent struct { Timestamp string `json:"timestamp" avro:"timestamp"` // ISO8601 timestamp UserAgent *string `json:"user_agent" avro:"user_agent"` // Browser user agent (nullable) IPHash *string `json:"ip_hash" avro:"ip_hash"` // Hashed IP for privacy (nullable) + // ClientIP is the source network, not the address: the edge zeroes the + // last IPv4 octet (or everything past the first 48 IPv6 bits) before + // publishing, so what the bus retains cannot single out a host. The + // consumer resolves it to a location and does not store it. Nullable and + // absent from events written before the field existed. + ClientIP *string `json:"client_ip" avro:"client_ip"` } diff --git a/internal/infrastructure/db/migrations/000125_engagement_origin.down.sql b/internal/infrastructure/db/migrations/000125_engagement_origin.down.sql new file mode 100644 index 00000000..288652a3 --- /dev/null +++ b/internal/infrastructure/db/migrations/000125_engagement_origin.down.sql @@ -0,0 +1,12 @@ +DROP TABLE email_opens; +ALTER TABLE email_link_clicks + DROP COLUMN announce_claimed_at, + DROP COLUMN announce_pending, + DROP COLUMN client, + DROP COLUMN device_type, + DROP COLUMN os, + DROP COLUMN browser, + DROP COLUMN browser_version, + DROP COLUMN country_code, + DROP COLUMN region, + DROP COLUMN city; diff --git a/internal/infrastructure/db/migrations/000125_engagement_origin.up.sql b/internal/infrastructure/db/migrations/000125_engagement_origin.up.sql new file mode 100644 index 00000000..dfbf97a3 --- /dev/null +++ b/internal/infrastructure/db/migrations/000125_engagement_origin.up.sql @@ -0,0 +1,47 @@ +-- Where an open or click came from: the mail client or image proxy, the +-- browser, device and operating system parsed from the user agent, and the +-- country, region and city resolved from the source network. Clicks gain +-- these columns on the per-link log. Opens get their own log, because the +-- progress row keeps only the first open per step and a second open from +-- another device is worth seeing. +ALTER TABLE email_link_clicks + ADD COLUMN client text NOT NULL DEFAULT '', + ADD COLUMN device_type text NOT NULL DEFAULT '', + ADD COLUMN os text NOT NULL DEFAULT '', + ADD COLUMN browser text NOT NULL DEFAULT '', + ADD COLUMN browser_version text NOT NULL DEFAULT '', + ADD COLUMN country_code text NOT NULL DEFAULT '', + ADD COLUMN region text NOT NULL DEFAULT '', + ADD COLUMN city text NOT NULL DEFAULT '', + -- A person's click waits out the burst window before its effects fire; + -- the row is the durable record of that pending work, so a consumer + -- restart inside the window loses nothing. The flag clears only once + -- the effects ran; a claim leases the row for the attempt, and a lease + -- that expires without completion is retried. + ADD COLUMN announce_pending boolean NOT NULL DEFAULT false, + ADD COLUMN announce_claimed_at timestamp with time zone; + +CREATE TABLE email_opens ( + id uuid PRIMARY KEY, + task_id uuid NOT NULL, + campaign_id uuid NOT NULL REFERENCES campaigns(id) ON DELETE CASCADE, + contact_id uuid NOT NULL REFERENCES contacts(id) ON DELETE CASCADE, + sequence_id uuid NOT NULL REFERENCES sequences(id) ON DELETE CASCADE, + opened_at timestamp with time zone NOT NULL, + machine boolean NOT NULL DEFAULT false, + machine_reason text NOT NULL DEFAULT '' CHECK (machine_reason IN ('', 'prefetch', 'instant')), + user_agent text NOT NULL DEFAULT '', + ip_hash text NOT NULL DEFAULT '', + client text NOT NULL DEFAULT '', + device_type text NOT NULL DEFAULT '', + os text NOT NULL DEFAULT '', + browser text NOT NULL DEFAULT '', + browser_version text NOT NULL DEFAULT '', + country_code text NOT NULL DEFAULT '', + region text NOT NULL DEFAULT '', + city text NOT NULL DEFAULT '' +); + +CREATE INDEX idx_email_opens_contact ON email_opens (contact_id, opened_at DESC); +CREATE INDEX idx_email_opens_campaign ON email_opens (campaign_id, opened_at DESC); +CREATE INDEX idx_email_opens_step ON email_opens (campaign_id, contact_id, sequence_id); diff --git a/internal/infrastructure/db/migrations/000126_link_clicks_pending_index.down.sql b/internal/infrastructure/db/migrations/000126_link_clicks_pending_index.down.sql new file mode 100644 index 00000000..0b87fd56 --- /dev/null +++ b/internal/infrastructure/db/migrations/000126_link_clicks_pending_index.down.sql @@ -0,0 +1 @@ +DROP INDEX CONCURRENTLY IF EXISTS idx_email_link_clicks_pending; diff --git a/internal/infrastructure/db/migrations/000126_link_clicks_pending_index.up.sql b/internal/infrastructure/db/migrations/000126_link_clicks_pending_index.up.sql new file mode 100644 index 00000000..b709119b --- /dev/null +++ b/internal/infrastructure/db/migrations/000126_link_clicks_pending_index.up.sql @@ -0,0 +1,4 @@ +-- The sweep for held-back click announcements reads only pending rows, so +-- the index is partial. Built concurrently, on its own, because the click +-- log is a live table and a plain CREATE INDEX would block writes to it. +CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_email_link_clicks_pending ON email_link_clicks (clicked_at) WHERE announce_pending; diff --git a/internal/infrastructure/pubsub/events.go b/internal/infrastructure/pubsub/events.go index 54bf1049..c1864be1 100644 --- a/internal/infrastructure/pubsub/events.go +++ b/internal/infrastructure/pubsub/events.go @@ -222,6 +222,14 @@ type TrackingEventPayload struct { // fetcher, a security gateway walking the links) so live views can badge // it instead of presenting it as a person's. Machine bool `json:"machine,omitempty"` + // OccurredAt is when the tracking service saw the open or click; the + // base timestamp is when this event was published. + OccurredAt time.Time `json:"occurred_at,omitempty"` + // Where and on what, from the engagement logs, for live feeds. + Client string `json:"client,omitempty"` + DeviceType string `json:"device_type,omitempty"` + CountryCode string `json:"country_code,omitempty"` + City string `json:"city,omitempty"` } // PageHitEvent is a website page view tied to a contact. diff --git a/internal/models/analytics.go b/internal/models/analytics.go index 8f2549a8..db5c9190 100644 --- a/internal/models/analytics.go +++ b/internal/models/analytics.go @@ -47,6 +47,27 @@ type CampaignAnalytics struct { Summary CampaignSummary `json:"summary"` Sequences []SequenceStats `json:"steps"` DailyStats []CampaignDailyStats `json:"daily_stats,omitempty"` + // Engagement is where and on what people opened and clicked, from the + // per-event logs. Human events only. + Engagement *CampaignEngagementBreakdown `json:"engagement,omitempty"` +} + +// EngagementBucket is one slice of a breakdown: how many distinct contacts +// opened and clicked from that country, client, or device. +type EngagementBucket struct { + Key string `json:"key"` + Opens int `json:"opens"` + Clicks int `json:"clicks"` +} + +// CampaignEngagementBreakdown is the "where from, on what" view of a +// campaign's human opens and clicks. Buckets are ordered by activity, capped, +// and keyed by ISO country code, client or browser name, and device type. +// Unknown is the empty key. +type CampaignEngagementBreakdown struct { + Countries []EngagementBucket `json:"countries"` + Clients []EngagementBucket `json:"clients"` + Devices []EngagementBucket `json:"devices"` } type CampaignSummary struct { diff --git a/internal/models/contact.go b/internal/models/contact.go index 9c5beb4f..597115cd 100644 --- a/internal/models/contact.go +++ b/internal/models/contact.go @@ -474,6 +474,10 @@ type ContactTimelineEvent struct { // logged per link (every click since link attribution shipped). Link *ContactLinkClick `json:"link,omitempty"` + // Where an email_opened / email_clicked event came from, when it was + // logged per event: mail client, device and rough location. + Origin *EngagementOrigin `json:"origin,omitempty"` + // Author (notes, lifecycle events). UserID *uuid.UUID `json:"user_id,omitempty"` } @@ -493,6 +497,27 @@ type ContactLinkClick struct { UserAgent string `json:"user_agent,omitempty"` } +// EngagementOrigin is what an open or click said about where it came from. +// Client names the mail client or image proxy when the user agent does +// (Gmail, Apple Mail, Outlook); the browser fields describe the rest. The +// location is resolved from the source network and the address itself is +// never stored. Every field is empty when unknown. +type EngagementOrigin struct { + Client string `json:"client,omitempty"` + DeviceType string `json:"device_type,omitempty"` + OS string `json:"os,omitempty"` + Browser string `json:"browser,omitempty"` + BrowserVersion string `json:"browser_version,omitempty"` + CountryCode string `json:"country_code,omitempty"` + Region string `json:"region,omitempty"` + City string `json:"city,omitempty"` +} + +// Empty reports whether nothing about the origin is known. +func (o EngagementOrigin) Empty() bool { + return o == EngagementOrigin{} +} + type ContactTimelineResult struct { Data []ContactTimelineEvent `json:"data"` // True if we hit the per-call cap and the caller should paginate diff --git a/internal/repository/pg_analytics.go b/internal/repository/pg_analytics.go index 2d14aaac..2e6541ad 100644 --- a/internal/repository/pg_analytics.go +++ b/internal/repository/pg_analytics.go @@ -18,6 +18,11 @@ type AnalyticsRepository interface { GetCampaignSummary(ctx context.Context, userID, campaignID uuid.UUID) (*models.CampaignSummary, *errx.Error) GetCampaignDailyStats(ctx context.Context, campaignID uuid.UUID, from, to time.Time) ([]models.CampaignDailyStats, *errx.Error) GetSequenceStats(ctx context.Context, campaignID uuid.UUID) ([]models.SequenceStats, *errx.Error) + // GetCampaignEngagementBreakdown groups the campaign's human opens and + // clicks by country, client and device: distinct contacts per bucket, the + // busiest `limit` buckets of each. A click counts as an open, as it does + // on the progress row. + GetCampaignEngagementBreakdown(ctx context.Context, campaignID uuid.UUID, limit int) (*models.CampaignEngagementBreakdown, *errx.Error) // Email account status GetAccountsWithErrors(ctx context.Context, userID uuid.UUID) ([]uuid.UUID, *errx.Error) @@ -180,6 +185,70 @@ func (r *analyticsRepository) GetCampaignDailyStats(ctx context.Context, campaig return stats, nil } +func (r *analyticsRepository) GetCampaignEngagementBreakdown(ctx context.Context, campaignID uuid.UUID, limit int) (*models.CampaignEngagementBreakdown, *errx.Error) { + if limit <= 0 { + limit = 8 + } + // One query per dimension over the union of both logs; the key + // expression is the only difference. The client falls back to the + // browser so a plain webmail open still lands in a named bucket, and + // unknown stays the empty key. + bucket := func(keyExpr string) ([]models.EngagementBucket, *errx.Error) { + query := ` + WITH ev AS ( + SELECT contact_id, 'open' AS kind, client, browser, device_type, country_code + FROM email_opens + WHERE campaign_id = $1 AND NOT machine + UNION ALL + SELECT contact_id, 'click' AS kind, client, browser, device_type, country_code + FROM email_link_clicks + WHERE campaign_id = $1 AND NOT machine + ) + SELECT ` + keyExpr + ` AS key, + COUNT(DISTINCT contact_id) AS opens, + COUNT(DISTINCT contact_id) FILTER (WHERE kind = 'click') AS clicks + FROM ev + GROUP BY 1 + ORDER BY opens + clicks DESC, key ASC + LIMIT $2 + ` + rows, err := r.DB.Query(ctx, query, campaignID, limit) + if err != nil { + db.CaptureError(err, query, []any{campaignID, limit}, "GetCampaignEngagementBreakdown") + return nil, errx.InternalError() + } + defer rows.Close() + out := []models.EngagementBucket{} + for rows.Next() { + var b models.EngagementBucket + if err := rows.Scan(&b.Key, &b.Opens, &b.Clicks); err != nil { + db.CaptureError(err, "", nil, "GetCampaignEngagementBreakdown scan") + return nil, errx.InternalError() + } + out = append(out, b) + } + if err := rows.Err(); err != nil { + db.CaptureError(err, query, []any{campaignID, limit}, "GetCampaignEngagementBreakdown rows") + return nil, errx.InternalError() + } + return out, nil + } + + countries, xerr := bucket(`country_code`) + if xerr != nil { + return nil, xerr + } + clients, xerr := bucket(`COALESCE(NULLIF(client, ''), browser)`) + if xerr != nil { + return nil, xerr + } + devices, xerr := bucket(`CASE WHEN device_type = 'unknown' THEN '' ELSE device_type END`) + if xerr != nil { + return nil, xerr + } + return &models.CampaignEngagementBreakdown{Countries: countries, Clients: clients, Devices: devices}, nil +} + func (r *analyticsRepository) GetSequenceStats(ctx context.Context, campaignID uuid.UUID) ([]models.SequenceStats, *errx.Error) { query := ` SELECT diff --git a/internal/repository/pg_campaign_progress.go b/internal/repository/pg_campaign_progress.go index daf66127..17e45e90 100644 --- a/internal/repository/pg_campaign_progress.go +++ b/internal/repository/pg_campaign_progress.go @@ -435,10 +435,17 @@ func (r *campaignProgressRepository) RecordEmailOpened(ctx context.Context, camp } // RecordEmailClicked records that an email link was clicked +// RecordEmailClicked stamps a person's click. It also counts as an open: +// the person had the email in front of them whatever the pixel saw, so a +// client that blocks images no longer reads "clicked, not opened". The +// implied open shares the click's timestamp, which is how UnrecordEmailClicked +// tells it from a pixel open. func (r *campaignProgressRepository) RecordEmailClicked(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) error { query := ` UPDATE campaign_contact_progress - SET clicked_at = NOW() + SET clicked_at = NOW(), + opened_at = COALESCE(opened_at, NOW()), + opened_machine = false WHERE campaign_id = $1 AND contact_id = $2 AND sequence_id = $3 @@ -457,7 +464,14 @@ func (r *campaignProgressRepository) RecordEmailClicked(ctx context.Context, cam func (r *campaignProgressRepository) UnrecordEmailClicked(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) error { query := ` UPDATE campaign_contact_progress ccp - SET clicked_at = NULL + SET clicked_at = NULL, + opened_at = CASE + WHEN ccp.opened_at = ccp.clicked_at AND NOT EXISTS ( + SELECT 1 FROM email_opens o + WHERE o.campaign_id = $1 AND o.contact_id = $2 AND o.sequence_id = $3 AND o.machine = false + ) THEN NULL + ELSE ccp.opened_at + END WHERE ccp.campaign_id = $1 AND ccp.contact_id = $2 AND ccp.sequence_id = $3 @@ -726,10 +740,14 @@ func (r *campaignProgressRepository) GetCampaignRollingRates(ctx context.Context return out, err } -// GetContactProgress retrieves progress for a specific contact in a campaign +// GetContactProgress retrieves progress for a specific contact in a campaign. +// A machine open comes back as NULL: this feeds routing and instant actions, +// and an automated fetch is not intent (machine clicks never stamp at all). func (r *campaignProgressRepository) GetContactProgress(ctx context.Context, campaignID, contactID uuid.UUID) ([]CampaignContactProgress, error) { query := ` - SELECT campaign_id, contact_id, sequence_id, sent_at, opened_at, clicked_at, replied_at, bounced_at, complained_at, COALESCE(reply_class, ''), COALESCE(ai_label, '') + SELECT campaign_id, contact_id, sequence_id, sent_at, + CASE WHEN opened_machine THEN NULL ELSE opened_at END, + clicked_at, replied_at, bounced_at, complained_at, COALESCE(reply_class, ''), COALESCE(ai_label, '') FROM campaign_contact_progress WHERE campaign_id = $1 AND contact_id = $2 ORDER BY sent_at ASC @@ -938,7 +956,9 @@ func (r *campaignProgressRepository) FindNextRoutedPair(ctx context.Context, cam FROM campaign_leads cl JOIN contacts c ON c.id = cl.contact_id LEFT JOIN LATERAL ( - SELECT sequence_id, sent_at, opened_at, clicked_at, replied_at, reply_class, ai_label + SELECT sequence_id, sent_at, + CASE WHEN p.opened_machine THEN NULL ELSE p.opened_at END AS opened_at, + clicked_at, replied_at, reply_class, ai_label FROM campaign_contact_progress p WHERE p.campaign_id = $1 AND p.contact_id = cl.contact_id AND p.sent_at IS NOT NULL ORDER BY p.sent_at DESC LIMIT 1 @@ -1077,7 +1097,9 @@ func (r *campaignProgressRepository) RouteContact(ctx context.Context, campaignI FROM campaign_leads cl JOIN contacts c ON c.id = cl.contact_id LEFT JOIN LATERAL ( - SELECT sequence_id, sent_at, opened_at, clicked_at, replied_at, reply_class, ai_label + SELECT sequence_id, sent_at, + CASE WHEN p.opened_machine THEN NULL ELSE p.opened_at END AS opened_at, + clicked_at, replied_at, reply_class, ai_label FROM campaign_contact_progress p WHERE p.campaign_id = $1 AND p.contact_id = cl.contact_id AND p.sent_at IS NOT NULL ORDER BY p.sent_at DESC LIMIT 1 @@ -1440,7 +1462,9 @@ func (r *campaignProgressRepository) CountUndeliverableLeads(ctx context.Context FROM campaign_leads cl JOIN contacts c ON c.id = cl.contact_id LEFT JOIN LATERAL ( - SELECT sequence_id, sent_at, opened_at, clicked_at, replied_at, reply_class, ai_label + SELECT sequence_id, sent_at, + CASE WHEN p.opened_machine THEN NULL ELSE p.opened_at END AS opened_at, + clicked_at, replied_at, reply_class, ai_label FROM campaign_contact_progress p WHERE p.campaign_id = $1 AND p.contact_id = cl.contact_id AND p.sent_at IS NOT NULL ORDER BY p.sent_at DESC LIMIT 1 diff --git a/internal/repository/pg_contact.go b/internal/repository/pg_contact.go index 9bfc6b1f..c8fcca75 100644 --- a/internal/repository/pg_contact.go +++ b/internal/repository/pg_contact.go @@ -2798,15 +2798,16 @@ func (r *contactRepository) GetDetail(ctx context.Context, userID uuid.UUID, org // 2. Engagement aggregates. campaign_contact_progress is the canonical // sent/opened/clicked/replied/bounced ledger keyed by (campaign, // contact, sequence). Counts come from non-null timestamp columns, - // "last X" comes from MAX() of each. + // "last X" comes from MAX() of each. Opens count people only: a machine + // open is a delivery signal, not engagement, here as in analytics. engQuery := ` SELECT COUNT(*) FILTER (WHERE sent_at IS NOT NULL) AS sent, - COUNT(*) FILTER (WHERE opened_at IS NOT NULL) AS opened, + COUNT(*) FILTER (WHERE opened_at IS NOT NULL AND NOT opened_machine) AS opened, COUNT(*) FILTER (WHERE clicked_at IS NOT NULL) AS clicked, COUNT(*) FILTER (WHERE replied_at IS NOT NULL) AS replied, COUNT(*) FILTER (WHERE bounced_at IS NOT NULL) AS bounced, - MAX(sent_at), MAX(opened_at), MAX(clicked_at), MAX(replied_at), MAX(bounced_at) + MAX(sent_at), MAX(opened_at) FILTER (WHERE NOT opened_machine), MAX(clicked_at), MAX(replied_at), MAX(bounced_at) FROM campaign_contact_progress WHERE contact_id = $1 ` @@ -3020,6 +3021,11 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, 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 @@ -3061,12 +3067,12 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, } for prows.Next() { var sentAt, openedAt, clickedAt, repliedAt, bouncedAt *time.Time - var openedMachine, hasLinkClicks bool + var openedMachine, hasOpenLog, hasLinkClicks bool var campID, seqID, eaID *uuid.UUID var campName, seqName, seqSubject, eaEmail, eaName *string if err := prows.Scan( &sentAt, &openedAt, &clickedAt, &repliedAt, &bouncedAt, - &openedMachine, &hasLinkClicks, + &openedMachine, &hasOpenLog, &hasLinkClicks, &campID, &campName, &seqID, &seqName, &seqSubject, &eaID, &eaEmail, &eaName, @@ -3101,7 +3107,13 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, events = append(events, ev) } makeEvent(sentAt, models.TimelineEmailSent) - makeEvent(openedAt, models.TimelineEmailOpened) + // 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 !hasLinkClicks { makeEvent(clickedAt, models.TimelineEmailClicked) } @@ -3113,7 +3125,8 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, // 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. clickQuery := ` - SELECT lc.id, lc.clicked_at, lc.destination, lc.label, lc.user_agent, lc.machine, lc.machine_reason, + SELECT lc.id, lc.task_id, lc.clicked_at, lc.destination, lc.label, lc.user_agent, lc.machine, lc.machine_reason, + lc.client, lc.device_type, lc.os, lc.browser, lc.browser_version, lc.country_code, lc.region, lc.city, cam.id, cam.name, seq.id, seq.name, seq.subject, ea.id, ea.email, ea.name @@ -3132,20 +3145,24 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, ORDER BY lc.clicked_at DESC LIMIT $4 ` - crows, err := r.DB.Query(ctx, clickQuery, contactID, userID, bound, limit) + crows, err := r.DB.Query(ctx, clickQuery, contactID, userID, bound, limit+1) if err != nil { db.CaptureError(err, clickQuery, []any{contactID, userID, bound, limit}, "ListTimeline link clicks") return nil, errx.InternalError() } for crows.Next() { var link models.ContactLinkClick + var origin models.EngagementOrigin + var taskID uuid.UUID var at time.Time var machine bool var reason string var campID, seqID, eaID *uuid.UUID var campName, seqName, seqSubject, eaEmail, eaName *string if err := crows.Scan( - &link.ID, &at, &link.URL, &link.Label, &link.UserAgent, &machine, &reason, + &link.ID, &taskID, &at, &link.URL, &link.Label, &link.UserAgent, &machine, &reason, + &origin.Client, &origin.DeviceType, &origin.OS, &origin.Browser, &origin.BrowserVersion, + &origin.CountryCode, &origin.Region, &origin.City, &campID, &campName, &seqID, &seqName, &seqSubject, &eaID, &eaEmail, &eaName, @@ -3167,6 +3184,11 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, SequenceName: seqName, Machine: &machine, Link: &link, + TaskID: &taskID, + } + if !origin.Empty() { + o := origin + ev.Origin = &o } if seqSubject != nil && *seqSubject != "" { ev.Subject = seqSubject @@ -3178,6 +3200,90 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, events = append(events, ev) } crows.Close() + if err := crows.Err(); err != nil { + db.CaptureError(err, clickQuery, nil, "ListTimeline link clicks rows") + return nil, errx.InternalError() + } + + // 10. Per-event opens: each one with what it came from, machine ones + // labelled. Same campaign scope as the progress feed. + openQuery := ` + SELECT o.id, o.task_id, o.opened_at, o.user_agent, o.machine, o.machine_reason, + o.client, o.device_type, o.os, o.browser, o.browser_version, o.country_code, o.region, o.city, + cam.id, cam.name, + seq.id, seq.name, seq.subject, + ea.id, ea.email, ea.name + FROM email_opens o + JOIN campaigns cam ON cam.id = o.campaign_id + JOIN sequences seq ON seq.id = o.sequence_id + LEFT JOIN LATERAL ( + SELECT ea.id, ea.email, ea.name + FROM tasks t + JOIN email_accounts ea ON ea.id = t.email_account_id + WHERE t.id = o.task_id + ) 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 + ` + orows, err := r.DB.Query(ctx, openQuery, contactID, userID, bound, limit+1) + if err != nil { + db.CaptureError(err, openQuery, []any{contactID, userID, bound, limit}, "ListTimeline opens") + return nil, errx.InternalError() + } + for orows.Next() { + var id, taskID uuid.UUID + var origin models.EngagementOrigin + var at time.Time + var machine bool + var reason, userAgent string + var campID, seqID, eaID *uuid.UUID + var campName, seqName, seqSubject, eaEmail, eaName *string + if err := orows.Scan( + &id, &taskID, &at, &userAgent, &machine, &reason, + &origin.Client, &origin.DeviceType, &origin.OS, &origin.Browser, &origin.BrowserVersion, + &origin.CountryCode, &origin.Region, &origin.City, + &campID, &campName, + &seqID, &seqName, &seqSubject, + &eaID, &eaEmail, &eaName, + ); err != nil { + orows.Close() + db.CaptureError(err, "", nil, "ListTimeline opens scan") + return nil, errx.InternalError() + } + ev := models.ContactTimelineEvent{ + Type: models.TimelineEmailOpened, + At: at, + EmailAccountID: eaID, + EmailAccountEmail: eaEmail, + EmailAccountName: eaName, + CampaignID: campID, + CampaignName: campName, + SequenceID: seqID, + SequenceName: seqName, + Machine: &machine, + TaskID: &taskID, + } + if !origin.Empty() { + o := origin + ev.Origin = &o + } + if seqSubject != nil && *seqSubject != "" { + ev.Subject = seqSubject + } + if reason != "" { + r := reason + ev.MachineReason = &r + } + events = append(events, ev) + } + orows.Close() + if err := orows.Err(); err != nil { + db.CaptureError(err, openQuery, nil, "ListTimeline opens rows") + return nil, errx.InternalError() + } if orgID != nil { // 2. Reply intents (inbound replies with classification). diff --git a/internal/repository/pg_email_opens.go b/internal/repository/pg_email_opens.go new file mode 100644 index 00000000..efe98f89 --- /dev/null +++ b/internal/repository/pg_email_opens.go @@ -0,0 +1,99 @@ +package repository + +import ( + "context" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/warmbly/warmbly/internal/models" +) + +// EmailOpen is one recorded open of one step: when, from what client and +// where. The progress row keeps only the first open per step; this keeps +// them all, machine ones included and labelled, for the contact's timeline +// and the campaign's audience breakdown. +type EmailOpen struct { + ID uuid.UUID + TaskID uuid.UUID + CampaignID uuid.UUID + ContactID uuid.UUID + SequenceID uuid.UUID + OpenedAt time.Time + Machine bool + MachineReason string + UserAgent string + IPHash string + Origin models.EngagementOrigin +} + +// Machine-open reasons stored in email_opens.machine_reason. +const ( + EmailOpenReasonPrefetch = "prefetch" // a mail client prefetch or a fetch with no browser + EmailOpenReasonInstant = "instant" // arrived inside the machine window after dispatch +) + +// EmailOpenRepository is the per-event open log. Only the tracking consumer +// writes it. +type EmailOpenRepository interface { + Insert(ctx context.Context, open *EmailOpen) error + // HasHumanOpen reports whether a person's own open is on record for the + // step, which decides whether an open a click implied can be withdrawn. + HasHumanOpen(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) (bool, error) + // Cleanup deletes opens older than the retention window. + Cleanup(ctx context.Context, olderThanDays int) (int64, error) +} + +type emailOpenRepository struct { + db *pgxpool.Pool +} + +// NewEmailOpenRepository creates a new email open repository. +func NewEmailOpenRepository(db *pgxpool.Pool) EmailOpenRepository { + return &emailOpenRepository{db: db} +} + +func (r *emailOpenRepository) Insert(ctx context.Context, o *EmailOpen) error { + if o.ID == uuid.Nil { + o.ID = uuid.New() + } + if o.OpenedAt.IsZero() { + o.OpenedAt = time.Now() + } + query := ` + INSERT INTO email_opens + (id, task_id, campaign_id, contact_id, sequence_id, opened_at, + machine, machine_reason, user_agent, ip_hash, + client, device_type, os, browser, browser_version, country_code, region, city) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18) + ` + _, err := r.db.Exec(ctx, query, + o.ID, o.TaskID, o.CampaignID, o.ContactID, o.SequenceID, o.OpenedAt, + o.Machine, o.MachineReason, o.UserAgent, o.IPHash, + o.Origin.Client, o.Origin.DeviceType, o.Origin.OS, o.Origin.Browser, o.Origin.BrowserVersion, + o.Origin.CountryCode, o.Origin.Region, o.Origin.City, + ) + return err +} + +func (r *emailOpenRepository) HasHumanOpen(ctx context.Context, campaignID, contactID, sequenceID uuid.UUID) (bool, error) { + query := ` + SELECT EXISTS ( + SELECT 1 FROM email_opens + WHERE campaign_id = $1 AND contact_id = $2 AND sequence_id = $3 AND machine = false + ) + ` + var ok bool + err := r.db.QueryRow(ctx, query, campaignID, contactID, sequenceID).Scan(&ok) + return ok, err +} + +func (r *emailOpenRepository) Cleanup(ctx context.Context, olderThanDays int) (int64, error) { + tag, err := r.db.Exec(ctx, + `DELETE FROM email_opens WHERE opened_at < NOW() - $1 * INTERVAL '1 day'`, + olderThanDays) + if err != nil { + return 0, err + } + return tag.RowsAffected(), nil +} diff --git a/internal/repository/pg_link_clicks.go b/internal/repository/pg_link_clicks.go index 3a54872f..30c315c0 100644 --- a/internal/repository/pg_link_clicks.go +++ b/internal/repository/pg_link_clicks.go @@ -2,10 +2,13 @@ package repository import ( "context" + "errors" "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" + "github.com/warmbly/warmbly/internal/models" ) // LinkClick is one recorded click on one tracked link. Machine clicks (a @@ -25,6 +28,11 @@ type LinkClick struct { Machine bool MachineReason string ClickedAt time.Time + // Origin is what the click said about where it came from. + Origin models.EngagementOrigin + // AnnouncePending marks a person's click whose effects wait for the + // burst window; the row is the durable record of that work. + AnnouncePending bool } // Machine-click reasons stored in email_link_clicks.machine_reason. @@ -57,6 +65,20 @@ type LinkClickRepository interface { // identity when known (two links may share a destination); the // destination is the fallback for events from an older tracking build. HasHumanClickOn(ctx context.Context, taskID uuid.UUID, linkID *uuid.UUID, destination string) (bool, error) + // ClaimAnnounce leases a pending click's announcement for one attempt + // and reports the click's classification at that moment. claimed is + // false when another attempt holds a live lease or the announcement is + // done. A lease that expires without CompleteAnnounce is offered again. + ClaimAnnounce(ctx context.Context, id uuid.UUID) (claimed bool, machine bool, err error) + // CompleteAnnounce records that the click's effects ran, so neither the + // timer nor the sweep offers it again. + CompleteAnnounce(ctx context.Context, id uuid.UUID) error + // ListPendingAnnouncements returns clicks whose announcement is still + // pending, not under a live lease, and whose burst window closed before + // `before`: what a consumer restart or a failed attempt left behind. + ListPendingAnnouncements(ctx context.Context, before time.Time, limit int) ([]LinkClick, error) + // Cleanup deletes clicks older than the retention window. + Cleanup(ctx context.Context, olderThanDays int) (int64, error) } type linkClickRepository struct { @@ -78,16 +100,91 @@ func (r *linkClickRepository) Insert(ctx context.Context, c *LinkClick) error { query := ` INSERT INTO email_link_clicks (id, tracked_link_id, task_id, campaign_id, contact_id, sequence_id, - destination, label, user_agent, ip_hash, machine, machine_reason, clicked_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) + destination, label, user_agent, ip_hash, machine, machine_reason, clicked_at, + client, device_type, os, browser, browser_version, country_code, region, city, + announce_pending) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, + $14, $15, $16, $17, $18, $19, $20, $21, $22) ` _, err := r.db.Exec(ctx, query, c.ID, c.TrackedLinkID, c.TaskID, c.CampaignID, c.ContactID, c.SequenceID, c.Destination, c.Label, c.UserAgent, c.IPHash, c.Machine, c.MachineReason, c.ClickedAt, + c.Origin.Client, c.Origin.DeviceType, c.Origin.OS, c.Origin.Browser, c.Origin.BrowserVersion, + c.Origin.CountryCode, c.Origin.Region, c.Origin.City, + c.AnnouncePending, ) return err } +// announceLease is how long one attempt at a click's effects may take before +// the sweep is allowed to try again. +const announceLease = "2 minutes" + +func (r *linkClickRepository) ClaimAnnounce(ctx context.Context, id uuid.UUID) (bool, bool, error) { + query := ` + UPDATE email_link_clicks + SET announce_claimed_at = NOW() + WHERE id = $1 AND announce_pending + AND (announce_claimed_at IS NULL OR announce_claimed_at < NOW() - INTERVAL '` + announceLease + `') + RETURNING machine + ` + var machine bool + err := r.db.QueryRow(ctx, query, id).Scan(&machine) + if errors.Is(err, pgx.ErrNoRows) { + return false, false, nil + } + if err != nil { + return false, false, err + } + return true, machine, nil +} + +func (r *linkClickRepository) CompleteAnnounce(ctx context.Context, id uuid.UUID) error { + _, err := r.db.Exec(ctx, `UPDATE email_link_clicks SET announce_pending = false WHERE id = $1`, id) + return err +} + +func (r *linkClickRepository) ListPendingAnnouncements(ctx context.Context, before time.Time, limit int) ([]LinkClick, error) { + query := ` + SELECT id, tracked_link_id, task_id, campaign_id, contact_id, sequence_id, + destination, label, machine, machine_reason, clicked_at, + client, device_type, os, browser, browser_version, country_code, region, city + FROM email_link_clicks + WHERE announce_pending AND clicked_at < $1 + AND (announce_claimed_at IS NULL OR announce_claimed_at < NOW() - INTERVAL '` + announceLease + `') + ORDER BY clicked_at ASC + LIMIT $2 + ` + rows, err := r.db.Query(ctx, query, before, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var out []LinkClick + for rows.Next() { + var c LinkClick + if err := rows.Scan(&c.ID, &c.TrackedLinkID, &c.TaskID, &c.CampaignID, &c.ContactID, &c.SequenceID, + &c.Destination, &c.Label, &c.Machine, &c.MachineReason, &c.ClickedAt, + &c.Origin.Client, &c.Origin.DeviceType, &c.Origin.OS, &c.Origin.Browser, &c.Origin.BrowserVersion, + &c.Origin.CountryCode, &c.Origin.Region, &c.Origin.City); err != nil { + return nil, err + } + c.AnnouncePending = true + out = append(out, c) + } + return out, rows.Err() +} + +func (r *linkClickRepository) Cleanup(ctx context.Context, olderThanDays int) (int64, error) { + tag, err := r.db.Exec(ctx, + `DELETE FROM email_link_clicks WHERE clicked_at < NOW() - $1 * INTERVAL '1 day'`, + olderThanDays) + if err != nil { + return 0, err + } + return tag.RowsAffected(), nil +} + func (r *linkClickRepository) CountRecentOtherLinks(ctx context.Context, taskID uuid.UUID, ipHash string, linkID *uuid.UUID, destination string, since time.Time) (int, error) { // A row is "another link" when both sides have a ticket and they differ; // a row without a ticket (older tracking build) is compared by @@ -111,7 +208,7 @@ func (r *linkClickRepository) CountRecentOtherLinks(ctx context.Context, taskID func (r *linkClickRepository) MarkBurst(ctx context.Context, taskID uuid.UUID, ipHash string, since time.Time) (int64, error) { query := ` UPDATE email_link_clicks - SET machine = true, machine_reason = $4 + SET machine = true, machine_reason = $4, announce_pending = false WHERE task_id = $1 AND ip_hash = $2 AND clicked_at >= $3 diff --git a/tracking/src/config.rs b/tracking/src/config.rs index d8f97949..3f1f81c3 100644 --- a/tracking/src/config.rs +++ b/tracking/src/config.rs @@ -53,6 +53,10 @@ pub struct Config { /// Shared bearer token for the backend internal API (required; same /// INTERNAL_API_TOKEN the workers use). pub internal_api_token: String, + /// Secret the source-address token is keyed with. An unkeyed hash of an + /// IPv4 address is reversible by enumeration; a keyed one is only a + /// stable name for one source. Defaults to the internal API token. + pub ip_hash_key: String, /// Per-source request budget for both tracking endpoints (default 300/min). pub rate_limit_per_min: u32, /// Page-view ingest budget per source per minute. Lower than the pixel @@ -198,6 +202,11 @@ impl Config { trusted_proxies, client_ip_header ); + let ip_hash_key = env::var("TRACKING_IP_HASH_KEY") + .ok() + .filter(|v| !v.trim().is_empty()) + .unwrap_or_else(|| internal_api_token.clone()); + Ok(Self { env: env_name, host, @@ -213,6 +222,7 @@ impl Config { schema_registry_key, schema_registry_secret, backend_internal_url, + ip_hash_key, internal_api_token, rate_limit_per_min, pagehit_rate_limit_per_min, @@ -290,6 +300,10 @@ impl Config { schema_registry_key, schema_registry_secret, backend_internal_url, + ip_hash_key: env::var("TRACKING_IP_HASH_KEY") + .ok() + .filter(|v| !v.trim().is_empty()) + .unwrap_or_else(|| internal_api_token.clone()), internal_api_token, rate_limit_per_min: 300, pagehit_rate_limit_per_min: 60, diff --git a/tracking/src/events.rs b/tracking/src/events.rs index 7d38786d..fd6499b8 100644 --- a/tracking/src/events.rs +++ b/tracking/src/events.rs @@ -14,4 +14,7 @@ pub struct TrackingEvent { pub timestamp: String, pub user_agent: Option, pub ip_hash: Option, + /// The source network (last IPv4 octet zeroed, IPv6 cut to 48 bits), + /// enough for the consumer's location lookup without naming a host. + pub client_ip: Option, } diff --git a/tracking/src/handlers.rs b/tracking/src/handlers.rs index dabc868b..6e68b0d2 100644 --- a/tracking/src/handlers.rs +++ b/tracking/src/handlers.rs @@ -46,6 +46,8 @@ pub struct AppState { /// Proxies whose forwarded-IP header is believed, and which header pub trusted_proxies: Arc>, pub client_ip_header: Arc, + /// Key for the source-address token (see `hash_ip`) + pub ip_hash_key: Arc, } impl AppState { @@ -75,6 +77,7 @@ impl AppState { hit_rate_limiter: Arc::new(RateLimiter::new(config.pagehit_rate_limit_per_min)), trusted_proxies: Arc::new(config.trusted_proxies.clone()), client_ip_header: Arc::new(config.client_ip_header.clone()), + ip_hash_key: Arc::new(config.ip_hash_key.clone()), } } @@ -124,13 +127,15 @@ pub async fn track_open( return pixel_response(); } - // Extract IP hash for deduplication + rate limiting - let ip_hash = Some(hash_ip(&client_ip( + // The address is hashed for deduplication + rate limiting; only its + // network travels with the event, for the location lookup downstream. + let ip = client_ip( peer, &headers, &state.trusted_proxies, &state.client_ip_header, - ))); + ); + let ip_hash = Some(hash_ip(&state.ip_hash_key, &ip)); // Anti-flood: over-budget sources still get the pixel (real mail clients // must never see a broken image), but nothing is published. @@ -168,6 +173,7 @@ pub async fn track_open( timestamp: Utc::now().to_rfc3339(), user_agent, ip_hash, + client_ip: Some(anonymize_ip(&ip)).filter(|n| !n.is_empty()), }) .await; }); @@ -192,13 +198,15 @@ pub async fn track_click( return (StatusCode::NOT_FOUND, "Unknown link").into_response(); } - // Anti-flood: cap total request rate per source - let ip_hash = Some(hash_ip(&client_ip( + // Anti-flood: cap total request rate per source. Only the address's + // network rides along, for the location lookup downstream. + let ip = client_ip( peer, &headers, &state.trusted_proxies, &state.client_ip_header, - ))); + ); + let ip_hash = Some(hash_ip(&state.ip_hash_key, &ip)); let source = ip_hash.clone().unwrap_or_else(|| "unknown".to_string()); if !state.rate_limiter.allow(&source).await { return (StatusCode::TOO_MANY_REQUESTS, "Slow down").into_response(); @@ -257,6 +265,7 @@ pub async fn track_click( timestamp: Utc::now().to_rfc3339(), user_agent, ip_hash, + client_ip: Some(anonymize_ip(&ip)).filter(|n| !n.is_empty()), }) .await; }); @@ -320,7 +329,7 @@ pub async fn track_page_hit( &state.trusted_proxies, &state.client_ip_header, ); - let source = hash_ip(&ip); + let source = hash_ip(&state.ip_hash_key, &ip); // Anti-flood: page views have their own, tighter budget on top of the // shared one, and the shared one counts too so a flood here also @@ -467,9 +476,32 @@ fn client_ip( } } -fn hash_ip(ip: &str) -> String { - // Hash the IP for privacy +/// The network an address belongs to, for the location lookup downstream: +/// the last IPv4 octet zeroed, an IPv6 address cut to its first 48 bits. +/// City-level resolution survives; a single host is no longer named, so the +/// bus can retain the event without retaining the address. +fn anonymize_ip(ip: &str) -> String { + match ip.parse::() { + Ok(IpAddr::V4(v4)) => { + let o = v4.octets(); + format!("{}.{}.{}.0", o[0], o[1], o[2]) + } + Ok(IpAddr::V6(v6)) => { + let s = v6.segments(); + std::net::Ipv6Addr::new(s[0], s[1], s[2], 0, 0, 0, 0, 0).to_string() + } + Err(_) => String::new(), + } +} + +/// A stable, keyed token for a source address: the same source gets the same +/// token (dedupe, rate limits, the burst rule), and nobody holding the token +/// can enumerate IPv4 space to get the address back, because the key is +/// secret. The key goes in first so the digest is not one of a public value. +fn hash_ip(key: &str, ip: &str) -> String { let mut hasher = Sha256::new(); + hasher.update(key.as_bytes()); + hasher.update([0u8]); hasher.update(ip.as_bytes()); let result = hasher.finalize(); format!("{:x}", result)[..16].to_string() // Take first 16 chars @@ -536,6 +568,23 @@ mod tests { ); } + #[test] + fn hash_ip_is_keyed_and_stable() { + assert_eq!(hash_ip("k", "203.0.113.9"), hash_ip("k", "203.0.113.9")); + assert_ne!(hash_ip("k", "203.0.113.9"), hash_ip("other", "203.0.113.9")); + assert_eq!(hash_ip("k", "203.0.113.9").len(), 16); + } + + #[test] + fn anonymize_ip_keeps_only_the_network() { + assert_eq!(anonymize_ip("203.0.113.9"), "203.0.113.0"); + assert_eq!( + anonymize_ip("2001:db8:abcd:1234:5678::1"), + "2001:db8:abcd::" + ); + assert_eq!(anonymize_ip("not an ip"), ""); + } + #[test] fn host_of_strips_scheme_and_path() { assert_eq!(host_of("https://WWW.Example.com/a?b#c"), "www.example.com"); diff --git a/tracking/src/kafka.rs b/tracking/src/kafka.rs index b32160e4..7ff43b59 100644 --- a/tracking/src/kafka.rs +++ b/tracking/src/kafka.rs @@ -28,7 +28,8 @@ pub const TRACKING_EVENT_SCHEMA: &str = r#" {"name": "link_id", "type": ["null", "string"], "default": null}, {"name": "timestamp", "type": "string", "avro.java.string": "String"}, {"name": "user_agent", "type": ["null", "string"], "default": null}, - {"name": "ip_hash", "type": ["null", "string"], "default": null} + {"name": "ip_hash", "type": ["null", "string"], "default": null}, + {"name": "client_ip", "type": ["null", "string"], "default": null} ] } "#; @@ -81,6 +82,13 @@ impl ToAvroValue for TrackingEvent { None => Value::Union(0, Box::new(Value::Null)), }, ), + ( + "client_ip", + match &self.client_ip { + Some(net) => Value::Union(1, Box::new(Value::String(net.clone()))), + None => Value::Union(0, Box::new(Value::Null)), + }, + ), ] } } diff --git a/web/src/app/app/campaigns/[id]/page.tsx b/web/src/app/app/campaigns/[id]/page.tsx index 8da196ec..1cae09e5 100644 --- a/web/src/app/app/campaigns/[id]/page.tsx +++ b/web/src/app/app/campaigns/[id]/page.tsx @@ -8,6 +8,7 @@ import { } from "lucide-react"; import { useCampaign } from "@/hooks/context/campaign"; import useCampaignAnalytics from "@/lib/api/hooks/app/analytics/useCampaignAnalytics"; +import type { CampaignEngagementBreakdown, EngagementBucket } from "@/lib/api/models/app/analytics/CampaignAnalytics"; import useCampaignDailyStats from "@/lib/api/hooks/app/analytics/useCampaignDailyStats"; import { SectionBar, Stat, StatStrip } from "@/components/layout/Page"; import { MultiTrend, type TrendSeries } from "@/components/ui/charts"; @@ -271,6 +272,8 @@ export default function CampaignOverview() { )} + + {/* quick breakdown strip below sequence table, mobile-friendly summary */} @@ -329,3 +332,93 @@ export default function CampaignOverview() { ); } + +// Country names from the browser's own locale data; the code stays as the +// fallback for anything it does not know. +const REGION_NAMES = (() => { + try { + return new Intl.DisplayNames(undefined, { type: "region" }); + } catch { + return null; + } +})(); + +function countryName(code: string): string { + if (!code) return "Unknown"; + try { + return REGION_NAMES?.of(code.toUpperCase()) ?? code; + } catch { + return code; + } +} + +function bucketLabel(kind: "countries" | "clients" | "devices", key: string): string { + if (kind === "countries") return countryName(key); + if (!key) return "Unknown"; + if (kind === "devices") return key.charAt(0).toUpperCase() + key.slice(1); + return key; +} + +// Where and on what people opened and clicked: the busiest countries, mail +// clients and devices, from the per-event logs (human events only, so a +// security scanner's data centre never leads the list). +function EngagementAudience({ + breakdown, + loading, +}: { + breakdown: CampaignEngagementBreakdown | null; + loading: boolean; +}) { + const columns: { kind: "countries" | "clients" | "devices"; label: string; rows: EngagementBucket[] }[] = [ + { kind: "countries", label: "Country", rows: breakdown?.countries ?? [] }, + { kind: "clients", label: "Mail client", rows: breakdown?.clients ?? [] }, + { kind: "devices", label: "Device", rows: breakdown?.devices ?? [] }, + ]; + const empty = columns.every((c) => c.rows.length === 0); + return ( +
+ + {loading ? ( +
+ ) : empty ? ( +
+

No opens or clicks yet

+

+ Once people open and click, this shows which countries, mail clients and devices they did it from. +

+
+ ) : ( +
+ {columns.map((c) => ( +
+
+ {c.label} + Opens + Clicks +
+ {c.rows.length === 0 ? ( +
Nothing yet
+ ) : ( +
+ {c.rows.map((r) => ( +
+ + {bucketLabel(c.kind, r.key)} + + + {r.opens} + + + {r.clicks} + +
+ ))} +
+ )} +
+ ))} +
+ )} +
+ ); +} diff --git a/web/src/components/app/contacts/ContactsTable.tsx b/web/src/components/app/contacts/ContactsTable.tsx index 3bbe8a71..0bd5c8ae 100644 --- a/web/src/components/app/contacts/ContactsTable.tsx +++ b/web/src/components/app/contacts/ContactsTable.tsx @@ -21,6 +21,7 @@ import { ClockIcon, CornerUpLeftIcon, DownloadIcon, + InfoIcon, LayersIcon, Loader2Icon, MailIcon, @@ -958,7 +959,17 @@ function ContactsTableBody({ {embedded ? "Progress" : "Status"} {embedded && ( <> - Opened + + + Opened + + + + + Clicked Replied diff --git a/web/src/components/app/contacts/contact-edit/ActivityTab.tsx b/web/src/components/app/contacts/contact-edit/ActivityTab.tsx index 64618d3b..b4a1aea6 100644 --- a/web/src/components/app/contacts/contact-edit/ActivityTab.tsx +++ b/web/src/components/app/contacts/contact-edit/ActivityTab.tsx @@ -50,7 +50,10 @@ import { import useContactTimeline from "@/lib/api/hooks/app/contacts/useContactTimeline"; import useContactCampaignStates from "@/lib/api/hooks/app/contacts/useContactCampaignStates"; import type ContactTimelineEvent from "@/lib/api/models/app/contacts/ContactTimelineEvent"; -import type { ContactTimelineEventType } from "@/lib/api/models/app/contacts/ContactTimelineEvent"; +import type { + ContactTimelineEventType, + EngagementOrigin, +} from "@/lib/api/models/app/contacts/ContactTimelineEvent"; import type ContactCampaignState from "@/lib/api/models/app/contacts/ContactCampaignState"; import type { ContactCampaignStep, @@ -662,6 +665,9 @@ function applyFilters( e.link?.label, e.link?.utm_content, e.link?.utm_campaign, + e.origin?.client, + e.origin?.city, + e.origin?.country_code, ] .filter(Boolean) .join(" ") @@ -1112,6 +1118,14 @@ function detailsFor(e: ContactTimelineEvent): [string, React.ReactNode][] { add("UTM content", l.utm_content); add("Browser", l.user_agent); } + if (e.origin) { + const o = e.origin; + add("Client", o.client); + add("Device", cap(o.device_type ?? "")); + add("Operating system", o.os); + add("Browser", [o.browser, o.browser_version].filter(Boolean).join(" ")); + add("Location", [o.city, o.region, o.country_code].filter(Boolean).join(", ")); + } if (e.type === "email_opened" || e.type === "email_clicked") { add("Classified as", e.machine ? machineLabel(e.machine_reason) : "A person"); } @@ -1159,6 +1173,17 @@ function cap(s: string): string { return s.charAt(0).toUpperCase() + s.slice(1); } +// "Gmail", or "Chrome on Windows", or "Mobile" when the user agent said +// little; empty when it said nothing. +function originLabel(o: EngagementOrigin): string { + if (o.client) return o.client; + const browser = [o.browser, o.browser_version ? o.browser_version.split(".")[0] : ""].filter(Boolean).join(" "); + if (browser && o.os) return `${browser} on ${o.os}`; + if (browser) return browser; + if (o.os) return o.os; + return cap(o.device_type ?? ""); +} + function EventMeta({ event, highlight, @@ -1311,6 +1336,18 @@ function EventMeta({ , ); } + if (event.origin) { + const on = originLabel(event.origin); + if (on) parts.push({on}); + const where = [event.origin.city, event.origin.country_code].filter(Boolean).join(", "); + if (where) { + parts.push( + + + , + ); + } + } if (event.intent) { parts.push(intent: {event.intent}); } diff --git a/web/src/hooks/useCampaignChannel.ts b/web/src/hooks/useCampaignChannel.ts index 0bbc433d..18f9b61b 100644 --- a/web/src/hooks/useCampaignChannel.ts +++ b/web/src/hooks/useCampaignChannel.ts @@ -35,6 +35,35 @@ export interface ActivityItem { timestamp: Date; } +// Open/click payload (pubsub.TrackingEventPayload): who, whether a person +// did it, and what the logs say about client and location. +interface TrackingPayload { + contact_email?: string; + original_url?: string; + link_label?: string; + machine?: boolean; + // When the tracking service saw it; the envelope timestamp is the publish time. + occurred_at?: string; + timestamp?: string; + client?: string; + device_type?: string; + country_code?: string; + city?: string; +} + +// The edge's time when it is carried and parses; receipt time otherwise. +function occurredAt(p: TrackingPayload): Date { + const raw = p.occurred_at || p.timestamp; + const d = raw ? new Date(raw) : null; + return d && !Number.isNaN(d.getTime()) ? d : new Date(); +} + +// " (Gmail, Berlin DE)" or "" when the logs say nothing. +function whereFrom(p: TrackingPayload): string { + const bits = [p.client, [p.city, p.country_code].filter(Boolean).join(' ')].filter(Boolean); + return bits.length ? ` (${bits.join(', ')})` : ''; +} + // Hook return type export interface CampaignChannelState { isConnected: boolean; @@ -118,35 +147,30 @@ export function useCampaignChannel(campaignId: string): CampaignChannelState { break; } case 'EMAIL_OPENED': { - const data = payload as { contact_email?: string }; + const data = payload as TrackingPayload; addActivity({ id: nextId('open'), type: 'opened', contactEmail: data.contact_email || 'Unknown', - message: `Opened by ${data.contact_email || 'Unknown'}`, - timestamp: new Date(), + message: `${data.machine ? 'Auto-opened' : 'Opened'} by ${data.contact_email || 'Unknown'}${whereFrom(data)}`, + timestamp: occurredAt(data), }); break; } case 'EMAIL_CLICKED': { - const data = payload as { - contact_email?: string; - original_url?: string; - link_label?: string; - machine?: boolean; - }; - const who = data.contact_email || 'Unknown'; + const data = payload as TrackingPayload; + const who = `${data.contact_email || 'Unknown'}${whereFrom(data)}`; const target = data.link_label || data.original_url; addActivity({ id: nextId('click'), type: 'clicked', - contactEmail: who, + contactEmail: data.contact_email || 'Unknown', message: data.machine ? `Automated click on ${target ?? 'a link'} for ${who} (not counted)` : target ? `Click from ${who} → ${target}` : `Click from ${who}`, - timestamp: new Date(), + timestamp: occurredAt(data), }); break; } diff --git a/web/src/lib/api/models/app/analytics/CampaignAnalytics.ts b/web/src/lib/api/models/app/analytics/CampaignAnalytics.ts index 2a8fc357..796e20a6 100644 --- a/web/src/lib/api/models/app/analytics/CampaignAnalytics.ts +++ b/web/src/lib/api/models/app/analytics/CampaignAnalytics.ts @@ -41,6 +41,21 @@ export interface SequenceStats { bounces: number } +// One slice of the engagement breakdown: distinct contacts who opened and +// clicked from that country (ISO code), client or browser, or device type. +// An empty key is "unknown". +export interface EngagementBucket { + key: string + opens: number + clicks: number +} + +export interface CampaignEngagementBreakdown { + countries: EngagementBucket[] + clients: EngagementBucket[] + devices: EngagementBucket[] +} + export default interface CampaignAnalytics { campaign_id: string name: string @@ -49,4 +64,6 @@ export default interface CampaignAnalytics { summary: CampaignSummary steps: SequenceStats[] daily_stats?: DailyStats[] + // Where and on what people opened and clicked; human events only. + engagement?: CampaignEngagementBreakdown | null } diff --git a/web/src/lib/api/models/app/contacts/ContactTimelineEvent.ts b/web/src/lib/api/models/app/contacts/ContactTimelineEvent.ts index f3cf41bc..9ce1d48a 100644 --- a/web/src/lib/api/models/app/contacts/ContactTimelineEvent.ts +++ b/web/src/lib/api/models/app/contacts/ContactTimelineEvent.ts @@ -67,6 +67,20 @@ export interface ContactLinkClick { user_agent?: string; } +// Where an open or click came from, when it was logged per event: the mail +// client or image proxy when the user agent names one, otherwise browser, +// OS and device; the location resolved from the source network. +export interface EngagementOrigin { + client?: string; + device_type?: string; + os?: string; + browser?: string; + browser_version?: string; + country_code?: string; + region?: string; + city?: string; +} + export default interface ContactTimelineEvent { type: ContactTimelineEventType; at: string; @@ -80,6 +94,9 @@ export default interface ContactTimelineEvent { // Per-link detail behind an email_clicked event. link?: ContactLinkClick | null; + // Where an email_opened / email_clicked event came from. + origin?: EngagementOrigin | null; + email_account_id?: string | null; email_account_email?: string | null; email_account_name?: string | null;