From 27630eec0a5f56542e352783728ff3cc1091c8c9 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 29 Aug 2026 03:25:50 -0700 Subject: [PATCH] feat: add website visitor tracking for issue #255 with migration 000106 (website_tracking_settings, website_visitors, website_page_hits, all registered in the orgtransfer spec), a consent-gated dependency-free tracking.js served by the Rust tracking service with a rate-limited, size-capped, prefetch-filtered POST /p ingest that forwards to a new backend internal page-hits endpoint for server-side user-agent and GeoIP enrichment, contact identification only through the click ticket the redirect appends to registered hosts, a per-workspace retention job, page_hit events with an expandable detail view in the contact Activity timeline, a Settings > Website tracking page for the snippet and consent, location and retention configuration, realtime PAGE_HIT fanout, and a website tracking guide plus endpoint, export and configuration docs --- cmd/backend/main.go | 10 + deploy/config/env.example | 2 + docs/content/docs/api/endpoints.mdx | 6 + docs/content/docs/api/realtime.mdx | 2 +- .../docs/development/configuration.mdx | 1 + docs/content/docs/development/events.mdx | 2 + docs/content/docs/guides/meta.json | 1 + docs/content/docs/guides/website-tracking.mdx | 80 ++++ .../docs/guides/workspace-export-import.mdx | 4 +- internal/api/handler/handler.go | 4 + internal/api/handler/internal_tracked_link.go | 7 +- internal/api/handler/website_tracking.go | 123 +++++ internal/api/routes.go | 12 + internal/app/orgtransfer/spec.go | 22 + internal/app/websitetracking/service.go | 439 ++++++++++++++++++ internal/app/websitetracking/service_test.go | 53 +++ .../000106_website_tracking.down.sql | 3 + .../migrations/000106_website_tracking.up.sql | 115 +++++ internal/infrastructure/pubsub/events.go | 29 ++ internal/jobs/website_tracking_retention.go | 56 +++ internal/models/contact.go | 7 + internal/models/website_tracking.go | 161 +++++++ internal/repository/pg_contact.go | 54 +++ internal/repository/pg_website_tracking.go | 302 ++++++++++++ .../lib/realtime_web/channels/org_channel.ex | 4 +- tracking/src/config.rs | 14 + tracking/src/handlers.rs | 224 ++++++++- tracking/src/hits.rs | 342 ++++++++++++++ tracking/src/links.rs | 6 + tracking/src/main.rs | 16 +- tracking/static/tracking.js | 127 +++++ web/src/app/app/settings/layout.tsx | 2 + web/src/app/app/settings/tracking/page.tsx | 289 ++++++++++++ .../app/contacts/contact-edit/ActivityTab.tsx | 146 +++++- web/src/hooks/useRealtimeEvents.ts | 11 +- .../app/websitetracking/websiteTracking.ts | 32 ++ .../app/websitetracking/useWebsiteTracking.ts | 36 ++ .../app/contacts/ContactTimelineEvent.ts | 37 +- .../WebsiteTrackingSettings.ts | 46 ++ 39 files changed, 2802 insertions(+), 25 deletions(-) create mode 100644 docs/content/docs/guides/website-tracking.mdx create mode 100644 internal/api/handler/website_tracking.go create mode 100644 internal/app/websitetracking/service.go create mode 100644 internal/app/websitetracking/service_test.go create mode 100644 internal/infrastructure/db/migrations/000106_website_tracking.down.sql create mode 100644 internal/infrastructure/db/migrations/000106_website_tracking.up.sql create mode 100644 internal/jobs/website_tracking_retention.go create mode 100644 internal/models/website_tracking.go create mode 100644 internal/repository/pg_website_tracking.go create mode 100644 tracking/src/hits.rs create mode 100644 tracking/static/tracking.js create mode 100644 web/src/app/app/settings/tracking/page.tsx create mode 100644 web/src/lib/api/client/app/websitetracking/websiteTracking.ts create mode 100644 web/src/lib/api/hooks/app/websitetracking/useWebsiteTracking.ts create mode 100644 web/src/lib/api/models/app/websitetracking/WebsiteTrackingSettings.ts diff --git a/cmd/backend/main.go b/cmd/backend/main.go index b838e376..fd653f5a 100644 --- a/cmd/backend/main.go +++ b/cmd/backend/main.go @@ -95,6 +95,7 @@ import ( warmupapp "github.com/warmbly/warmbly/internal/app/warmup" "github.com/warmbly/warmbly/internal/app/warmupcontent" "github.com/warmbly/warmbly/internal/app/webhook" + "github.com/warmbly/warmbly/internal/app/websitetracking" "github.com/warmbly/warmbly/internal/app/worker" "github.com/warmbly/warmbly/internal/app/worker_orchestrator" "github.com/warmbly/warmbly/internal/config" @@ -157,6 +158,7 @@ func main() { var rateLimitService ratelimit.RateLimitService var sequenceService sequence.SequenceService var contactService contact.ContactService + var websiteTrackingService websitetracking.Service var socketService socket.SocketService var uniboxService unibox.UniboxService var cipherService cipher.CipherService @@ -1575,6 +1577,13 @@ func main() { auditRetentionScheduler := jobs.NewAuditRetentionScheduler(auditRetentionJob, 6*time.Hour) go auditRetentionScheduler.Start(ctx) + // Website tracking: the snippet's settings, the ingest path the tracking + // service forwards page views to, and the per-workspace retention sweep + // that keeps the window promised on the settings page. + websiteTrackingRepo := repository.NewWebsiteTrackingRepository(primaryDB.Pool) + websiteTrackingService = websitetracking.NewService(websiteTrackingRepo, geoloc, streamingPublisher) + go jobs.NewWebsiteTrackingRetentionJob(websiteTrackingRepo).Start(ctx, 6*time.Hour) + // Warmup content generator: tops the AI thread bank up toward the // admin-configured per-pool/segment targets. The internal cadence gate // honours the admin's cadence_hours; it no-ops when generation is @@ -1856,6 +1865,7 @@ func main() { EmailMessageMap: emailMessageMapForHandler, EmailSyncState: emailSyncStateRepository, TrackedLinks: trackedLinkRepository, + WebsiteTrackingService: websiteTrackingService, UserRepo: userRepoForHandler, OrgRepo: organizationRepoForHandler, AttachmentRepo: attachmentRepoForHandler, diff --git a/deploy/config/env.example b/deploy/config/env.example index 390c78ce..9eb015b9 100644 --- a/deploy/config/env.example +++ b/deploy/config/env.example @@ -228,6 +228,8 @@ TRACKING_PORT=3000 # Resolves opaque /c/ click tickets via the backend internal API. # BACKEND_INTERNAL_URL=http://backend:8080 TRACKING_RATE_LIMIT_PER_MIN=300 +# Website page views accepted per source per minute (the snippet posts to /p). +TRACKING_PAGEHIT_RATE_LIMIT_PER_MIN=60 # === Realtime service (Elixir/Phoenix) === PHX_HOST=localhost diff --git a/docs/content/docs/api/endpoints.mdx b/docs/content/docs/api/endpoints.mdx index 05fdade1..18256e9e 100644 --- a/docs/content/docs/api/endpoints.mdx +++ b/docs/content/docs/api/endpoints.mdx @@ -295,6 +295,7 @@ These never accept an API key. They depend on a human-bound session: billing flo - `GET /me/danger-zone`, `POST /me/danger-zone/delete`, `DELETE /me/danger-zone/delete` - `GET /invitations`, `POST /invitations/accept` - All of `/organization/*` (create, switch, members, invitations, transfer ownership, avatar, danger zone) +- `GET /website-tracking/settings`, `PATCH /website-tracking/settings`, `POST /website-tracking/settings/rotate-key` (the [website tracking](/guides/website-tracking/) snippet's consent mode, location precision, allowed hosts and retention; JWT permission `MANAGE_SETTINGS`. The rotate is bodyless and safe to repeat, each call issues a new key) - All of `/subscription/*` (checkout, portal, cancel, change-plan, preview-change, enterprise-inquiry, discounts, referrals, etc.) - All of `/admin/*` @@ -372,6 +373,11 @@ External MCP servers whose tools the assistant can use (see [Connect MCP tools]( - `POST /oauth/register` (OAuth dynamic client registration, RFC 7591; open and per-IP rate-limited) - `GET /.well-known/oauth-authorization-server`, `GET /.well-known/oauth-protected-resource` (OAuth discovery metadata) +On the tracking service (the `TRACKING_DOMAIN` host, not the API), also public and rate-limited per source: + +- `GET /t/o/:task_id.png` (open pixel), `GET /c/:link_id` (click redirect) +- `GET /tracking.js` (the website tracking snippet), `POST /p` (page-view ingest; JSON body up to 8 KB, `429` over budget, `204` otherwise. Nothing in the request can name a contact) + ## Notes - When a route has both a JWT permission and an API permission listed, the dual-auth middleware checks the JWT user's organization role for browser callers and the key's permission bitmask for API key callers. They're independent gates: a user's role doesn't constrain what an API key can do beyond what was granted at creation. diff --git a/docs/content/docs/api/realtime.mdx b/docs/content/docs/api/realtime.mdx index 3a6041b8..c9977b93 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`. 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). `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/development/configuration.mdx b/docs/content/docs/development/configuration.mdx index b386a405..88ca0d68 100644 --- a/docs/content/docs/development/configuration.mdx +++ b/docs/content/docs/development/configuration.mdx @@ -383,6 +383,7 @@ The Rust open and click service. It reads its own environment, so these have to | `BACKEND_INTERNAL_URL` | Where tracking resolves opaque `/c/` click tickets. **Required**: the service exits at boot without it | none | | `INTERNAL_API_TOKEN` | Bearer token for that lookup. **Required**: the service exits at boot on an empty value | none | | `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` | | `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` | | `KAFKA_TRACKING_TOPIC` | Event topic, read by the Rust publisher **and** the Go subscriber | `tracking-events` | diff --git a/docs/content/docs/development/events.mdx b/docs/content/docs/development/events.mdx index 77a6db71..83c04b8b 100644 --- a/docs/content/docs/development/events.mdx +++ b/docs/content/docs/development/events.mdx @@ -82,6 +82,8 @@ Produced by the Rust tracking service when a pixel loads or a tracked link is cl `event_type` is `EMAIL_OPENED` or `EMAIL_CLICKED`; `original_url` is set only for clicks; IPs are stored as hashes, never raw. The struct is `events.TrackingEvent` in `internal/events/schemas.go`, mirrored in `tracking/src/events.rs`. +Website page views do not ride this topic. The tracking service forwards each accepted view to the backend's internal API (`POST /api/v1/internal/page-hits`) instead, because the backend is where the user agent and IP are turned into device and location, and the IP must not sit in a durable stream on the way there. + ## Analytics events The backend records sends on two write-only streams for downstream analytics: `email-events` carries `EMAIL_SENT` records (task, account, campaign, contact, message ID, recipient), and `warmup-events` carries `WARMUP_EMAIL_SENT` records (sender and target accounts, whether it was a reply). Nothing in the repo consumes them today; they exist so an external pipeline can tap the stream. diff --git a/docs/content/docs/guides/meta.json b/docs/content/docs/guides/meta.json index a0cadc93..894c22c9 100644 --- a/docs/content/docs/guides/meta.json +++ b/docs/content/docs/guides/meta.json @@ -14,6 +14,7 @@ "advisor", "---Contacts and inbox---", "contacts-crm", + "website-tracking", "unibox", "meetings", "---Automation---", diff --git a/docs/content/docs/guides/website-tracking.mdx b/docs/content/docs/guides/website-tracking.mdx new file mode 100644 index 00000000..24f08c59 --- /dev/null +++ b/docs/content/docs/guides/website-tracking.mdx @@ -0,0 +1,80 @@ +--- +title: "Website tracking" +description: "Install the tracking snippet, see which pages a contact visits, and control consent and retention." +--- + +Website tracking adds page views on your own site to a contact's activity timeline: which pages they landed on, where they came from, which campaign or UTM brought them, and what device they used. It is off until you turn it on, and the consent mode, location precision and retention window are yours to set. + +## Installation + +Open **Settings > Website tracking**, switch on **Record website visits**, add the hosts your site runs on, and paste the snippet before the closing `` tag on every page: + +```html + + +``` + +The settings page renders the exact snippet for your workspace, with your tracking host and site key filled in. The first line is a small stub so your page can call `warmbly()` before the script has loaded; the calls are queued and run once it has. + +The script is around 3 KB, has no dependencies, and reports through a single request per page view. Single-page applications are covered: a navigation that pushes a history entry counts as a page view, and so does the back button. + + +It only says which workspace a page view belongs to. If a copy of the snippet ends up somewhere it should not be, rotate the key from the settings page; the old key stops working at once and every installed snippet has to be updated. + + +### Hosts + +The **Your website hosts** list does two things. Page views reported from any other host are ignored, so nobody can point your site key at a site you do not run. And links in your campaign emails only identify a visitor when they lead to one of these hosts, so the identification ticket is never handed to a third-party site. A host covers its subdomains: `example.com` admits `www.example.com` and `app.example.com`. + +Leave the list empty and page views from any host are recorded, but none are ever tied to a contact. + +## How a visitor becomes a contact + +Every link in a campaign email is already a [tracked ticket](/guides/campaigns/) that resolves server-side to the recipient. When a recipient clicks one and the destination is one of your hosts, the redirect appends that ticket to the landing URL as `wbly_t`. The snippet reads it, removes it from the address bar immediately, and sends it with the page view. The server checks that the ticket belongs to a campaign in the same workspace as the site key and, only then, ties the browser to that contact. + +That is the only way a visit reaches a contact. There is no `identify(email)` call, because a public endpoint that accepted an email address would let anyone write page views into anyone's timeline. A ticket is unguessable and only ever in the hands of the person the email went to. + +Once a browser is tied to a contact, the page views it reported before identification, back to the start of the retention window, appear on that contact's timeline too. Views after it are attributed as they arrive. + +### Two people on one browser + +If a second contact's ticket arrives on a browser already tied to someone else, the records are split rather than merged: the earlier contact keeps the history up to that click, the browser gets a fresh visitor id, and everything from then on belongs to the second contact. Neither inherits the other's pages. + +On a shared device, call `warmbly('reset')` when it changes hands to drop the visitor and session ids. + +## Consent + +Choose the mode on the settings page: + +- **Ask first** (the default): the snippet stores nothing and sends nothing until your page calls `warmbly('consent', 'granted')`, typically from your cookie banner once the visitor accepts. The grant is remembered in the browser, so it is asked once. `warmbly('consent', 'denied')` withdraws it and clears the visitor id. +- **Record on load**: page views are recorded as soon as the page loads. Choose this only where you have a lawful basis without a prior opt-in. + +The mode is enforced on the server as well as in the snippet, so a stale snippet cannot downgrade a workspace that requires consent. In both modes a browser that sends Global Privacy Control or Do Not Track is never recorded. + +## What is collected + +| Reported by the browser | Derived on the server | Never kept | +|---|---|---| +| Page URL and title | Device type, operating system, browser and version, device brand (from the request's user agent) | The IP address | +| Referrer and referring domain | Country, and optionally region and city (from the IP address, at the precision you choose) | Anything typed on the page | +| UTM source, medium, campaign, term and content | | Email addresses or names | +| Language, timezone and screen resolution | | | +| Visitor and session ids | | | + +Nothing about the device or location is trusted from the page: the snippet does not send them, and a request that claimed them would be ignored. Set **Location from IP address** to **Do not keep** to drop location entirely. + +A session ends after 30 minutes without a page view. The first view of a session is marked as the landing page. + +## Retention + +**Keep visits for** sets how long page views are kept, from 7 to 365 days (90 by default). A job on the backend prunes each workspace's views past its own window every few hours; the promise on the settings page is what the job keeps. Deleting a contact deletes the browsing history tied to them at the same time. + +Page views, and the browser records that carry them, travel in the **Delivery events** group of a [workspace archive](/guides/workspace-export-import/). The tracking settings, including the site key, are part of the workspace itself so installed snippets keep reporting after a move. + +## In the timeline + +Page views show in the **Activity** tab of a contact under the **Website** filter, one line each with the page title or path, the referring domain, the device and any UTM source. Open a row for the full detail: URL, referrer, operating system, browser, resolution, language, timezone, location and every UTM parameter. New views for a contact appear live for everyone with the contact open. + +## Abuse controls + +The ingest endpoint on the tracking service is public, so it runs behind the same controls as the open pixel and click redirects: a per-source request budget (with a tighter one for page views on top), prefetch and crawler filtering, an 8 KB body cap with limits on every field, a short window that ignores reloads and double-fires, a per-source budget for unknown site keys, and a circuit breaker toward the backend. Over budget requests get `429`; everything else that is declined is acknowledged quietly so a probe learns nothing about which keys exist. diff --git a/docs/content/docs/guides/workspace-export-import.mdx b/docs/content/docs/guides/workspace-export-import.mdx index 693caeb7..76f7cf11 100644 --- a/docs/content/docs/guides/workspace-export-import.mdx +++ b/docs/content/docs/guides/workspace-export-import.mdx @@ -13,7 +13,7 @@ The data is split into groups. Every export includes **Workspace**; the rest are | Group | Contents | |-------|----------| -| Workspace | The organization, members, roles, teams, mailboxes, API keys, webhooks, and settings. Always included | +| Workspace | The organization, members, roles, teams, mailboxes, API keys, webhooks, and settings, including the website tracking site key. Always included | | Contacts | Contacts, categories, notes, activities, and the suppression list | | Campaigns | Campaigns, sequences, senders, attachments, and per-campaign settings | | CRM | Pipelines, deals, tasks, and meeting bookings | @@ -22,7 +22,7 @@ The data is split into groups. Every export includes **Workspace**; the rest are | Warmup | Warmup participation, routing rules, statistics, and appeals | | Inbox | Unified inbox threads, message bodies, and mailbox sync state | | Send history | Queued and completed send tasks with their payloads | -| Delivery events | Bounces, complaints, opens, clicks, and placement tests | +| Delivery events | Bounces, complaints, opens, clicks, placement tests, and website page views with the browser records that tie them to contacts | | Logs | Audit log, campaign logs, and notifications | | Billing history | Subscription, credit ledger, and referral records | diff --git a/internal/api/handler/handler.go b/internal/api/handler/handler.go index 869100f1..10c6558a 100644 --- a/internal/api/handler/handler.go +++ b/internal/api/handler/handler.go @@ -59,6 +59,7 @@ import ( "github.com/warmbly/warmbly/internal/app/warmup" "github.com/warmbly/warmbly/internal/app/warmupcontent" "github.com/warmbly/warmbly/internal/app/webhook" + "github.com/warmbly/warmbly/internal/app/websitetracking" "github.com/warmbly/warmbly/internal/app/worker" "github.com/warmbly/warmbly/internal/app/worker_orchestrator" "github.com/warmbly/warmbly/internal/pkg/generation" @@ -160,6 +161,9 @@ type Handler struct { // Advanced outreach controls AdvancedService advanced.Service + // Website tracking snippet: settings and the page-view ingest path. + WebsiteTrackingService websitetracking.Service + // Pre-send email verification (control-plane SMTP RCPT probe / pluggable // paid backend). Drops hard-bouncing addresses before a worker sends. EmailVerifyService emailverifyapp.Service diff --git a/internal/api/handler/internal_tracked_link.go b/internal/api/handler/internal_tracked_link.go index 33a93dde..e3fde919 100644 --- a/internal/api/handler/internal_tracked_link.go +++ b/internal/api/handler/internal_tracked_link.go @@ -11,7 +11,11 @@ import ( // Auth via middleware.InternalAuthMiddleware (INTERNAL_API_TOKEN, both sides). // // GET /api/v1/internal/tracked-links/:id -// -> 200 {"destination":"https://...","task_id":""} | 404 +// -> 200 {"destination":"https://...","task_id":"","identify":bool} | 404 +// +// identify is true when the workspace runs website tracking and registered +// the destination's host, so the redirect may append the ticket for the +// snippet to tie the browser to the contact. // // The tracking service caches positives and negatives aggressively and rate // limits miss-heavy sources before calling here, so this stays a cheap @@ -36,5 +40,6 @@ func (h *Handler) InternalGetTrackedLink(c *gin.Context) { c.JSON(http.StatusOK, gin.H{ "destination": link.Destination, "task_id": link.TaskID.String(), + "identify": h.websiteIdentifyForLink(c, link.CampaignID, link.Destination), }) } diff --git a/internal/api/handler/website_tracking.go b/internal/api/handler/website_tracking.go new file mode 100644 index 00000000..91a048b2 --- /dev/null +++ b/internal/api/handler/website_tracking.go @@ -0,0 +1,123 @@ +package handler + +import ( + "errors" + "net/http" + + "github.com/gin-gonic/gin" + "github.com/google/uuid" + "github.com/warmbly/warmbly/internal/api/middleware" + "github.com/warmbly/warmbly/internal/app/websitetracking" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/models" +) + +// Website tracking settings (JWT only, MANAGE_SETTINGS): the snippet's site +// key, consent mode, location precision, allowed hosts and retention. + +func (h *Handler) GetWebsiteTrackingSettings(c *gin.Context) { + orgID := middleware.GetOrganizationID(c) + if orgID == nil { + errx.JSON(c, errx.New(errx.BadRequest, "no organization selected")) + return + } + settings, xerr := h.WebsiteTrackingService.GetSettings(c.Request.Context(), *orgID) + if xerr != nil { + errx.JSON(c, xerr) + return + } + c.JSON(http.StatusOK, settings) +} + +func (h *Handler) UpdateWebsiteTrackingSettings(c *gin.Context) { + orgID := middleware.GetOrganizationID(c) + if orgID == nil { + errx.JSON(c, errx.New(errx.BadRequest, "no organization selected")) + return + } + userID, err := middleware.GetUserUUID(c) + if err != nil { + errx.JSON(c, errx.New(errx.BadRequest, "invalid user id")) + return + } + var req models.UpdateWebsiteTrackingSettingsRequest + if err := c.ShouldBindJSON(&req); err != nil { + errx.JSON(c, errx.ErrInvalid) + return + } + settings, xerr := h.WebsiteTrackingService.UpdateSettings(c.Request.Context(), *orgID, userID, &req) + if xerr != nil { + errx.JSON(c, xerr) + return + } + h.auditOrg(c, models.AuditActionUpdate, models.AuditEntitySettings, nil, nil, map[string]string{ + "scope": "website_tracking", + }) + c.JSON(http.StatusOK, settings) +} + +// RotateWebsiteTrackingKey issues a new site key. Snippets carrying the old +// one are refused from the next hit on. +func (h *Handler) RotateWebsiteTrackingKey(c *gin.Context) { + orgID := middleware.GetOrganizationID(c) + if orgID == nil { + errx.JSON(c, errx.New(errx.BadRequest, "no organization selected")) + return + } + userID, err := middleware.GetUserUUID(c) + if err != nil { + errx.JSON(c, errx.New(errx.BadRequest, "invalid user id")) + return + } + settings, xerr := h.WebsiteTrackingService.RotateSiteKey(c.Request.Context(), *orgID, userID) + if xerr != nil { + errx.JSON(c, xerr) + return + } + h.auditOrg(c, models.AuditActionUpdate, models.AuditEntitySettings, nil, nil, map[string]string{ + "scope": "website_tracking", + "action": "rotate_site_key", + }) + c.JSON(http.StatusOK, settings) +} + +// InternalIngestPageHit is where the tracking service forwards a page view +// after its own rate limiting, filtering and payload caps. +// +// POST /api/v1/internal/page-hits +// -> 204 accepted or quietly rejected | 200 {"new_visitor_key":...} +// -> 400 malformed | 404 unknown site key +// +// Unknown keys 404 so the edge can cache them negatively and cut off a +// probing source, exactly like an unknown click ticket. +func (h *Handler) InternalIngestPageHit(c *gin.Context) { + var req models.WebsiteHitRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid body"}) + return + } + result, err := h.WebsiteTrackingService.Ingest(c.Request.Context(), &req) + switch { + case errors.Is(err, websitetracking.ErrMalformed): + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid hit"}) + case errors.Is(err, websitetracking.ErrUnknownSite): + c.Status(http.StatusNotFound) + case errors.Is(err, websitetracking.ErrRejected): + c.Status(http.StatusNoContent) + case err != nil: + c.JSON(http.StatusInternalServerError, gin.H{"error": "ingest failed"}) + case result != nil && result.NewVisitorKey != "": + c.JSON(http.StatusOK, result) + default: + c.Status(http.StatusNoContent) + } +} + +// websiteIdentifyForLink is the click redirect's question: may this +// destination carry the visitor-identification ticket? +func (h *Handler) websiteIdentifyForLink(c *gin.Context, campaignID uuid.UUID, destination string) bool { + if h.WebsiteTrackingService == nil { + return false + } + return h.WebsiteTrackingService.ShouldIdentify(c.Request.Context(), campaignID, destination) +} diff --git a/internal/api/routes.go b/internal/api/routes.go index 90ea3e87..4c039eea 100644 --- a/internal/api/routes.go +++ b/internal/api/routes.go @@ -121,6 +121,11 @@ func Run( // here instead of touching Postgres (read-only, heavily cached there). internal.GET("/tracked-links/:id", h.InternalGetTrackedLink) + // Website page views: the tracking service forwards each counted hit + // here after its own rate limiting and filtering. Enrichment (user + // agent, IP location) and storage happen on this side. + internal.POST("/page-hits", h.InternalIngestPageHit) + // Worker mailbox-sync messageId -> internal email map (replaces the // former DynamoDB EmailMessageData table). Workers read/write it here. internal.GET("/email-message-map", h.InternalGetEmailMessageMap) @@ -1027,6 +1032,13 @@ func Run( // to silence the Advisor for a whole workspace. jwtOnly.PATCH("/advisor/settings", m.RequireOrganization(), m.RequirePermission(models.PermManageSettings), h.UpdateAdvisorSettings) + // Website tracking is privacy governance (consent mode, retention, + // what the snippet may collect), so it is JWT-only like the other + // org settings. The rotate is bodyless and idempotent per call. + jwtOnly.GET("/website-tracking/settings", m.RequireOrganization(), m.RequirePermission(models.PermManageSettings), h.GetWebsiteTrackingSettings) + jwtOnly.PATCH("/website-tracking/settings", m.RequireOrganization(), m.RequirePermission(models.PermManageSettings), h.UpdateWebsiteTrackingSettings) + jwtOnly.POST("/website-tracking/settings/rotate-key", m.RequireOrganization(), m.RequirePermission(models.PermManageSettings), h.RotateWebsiteTrackingKey) + // The agent fix is JWT-only for the same reason the dashboard agent // is: it acts as a named member, inside their permissions, and there // is no API scope that should let a key spend credits rewriting a diff --git a/internal/app/orgtransfer/spec.go b/internal/app/orgtransfer/spec.go index d48f132c..6f88967a 100644 --- a/internal/app/orgtransfer/spec.go +++ b/internal/app/orgtransfer/spec.go @@ -265,6 +265,20 @@ var Tables = []Table{ Name: "contact_research_runs", Group: models.OrgDataGroupContacts, Scope: scopeOrgAlt, }, + { + // Core rather than events: the site key is what the customer's own + // website already carries, so it has to survive the move or every + // installed snippet goes dark. + Name: "website_tracking_settings", Group: models.OrgDataGroupCore, + Scope: scopeOrg, + Note: "The site key travels so snippets already installed keep reporting once the tracking host follows.", + }, + { + // After contacts: contact_id is nullable, so a contacts-less run + // blanks it rather than aborting, and the visits arrive anonymous. + Name: "website_visitors", Group: models.OrgDataGroupEvents, + Scope: scopeOrg, + }, // ---------- campaigns ---------- { @@ -605,6 +619,14 @@ var Tables = []Table{ Name: "api_key_usage_logs", Group: models.OrgDataGroupEvents, Scope: `api_key_id IN ` + orgAPIKeys, }, + { + // Bounded by the workspace's own retention window, so even a busy + // site adds at most a year of rows; the events group is already the + // opt-out for volume. + Name: "website_page_hits", Group: models.OrgDataGroupEvents, + Scope: scopeOrg, + Note: "Page views collected under the workspace's consent setting. The retention sweep on the destination keeps pruning them by the imported window.", + }, // ---------- logs ---------- { diff --git a/internal/app/websitetracking/service.go b/internal/app/websitetracking/service.go new file mode 100644 index 00000000..c9e99be2 --- /dev/null +++ b/internal/app/websitetracking/service.go @@ -0,0 +1,439 @@ +// Package websitetracking is the control plane behind the website tracking +// snippet: the workspace's settings, and the ingest path the Rust tracking +// service forwards page views to. Device and location are derived here from +// the request facts the edge saw; nothing about them is trusted from the +// browser, and a hit only reaches a contact through a click ticket the +// contact's own email carried. +package websitetracking + +import ( + "context" + "crypto/rand" + "encoding/hex" + "errors" + "net" + "net/netip" + "net/url" + "strings" + "time" + + "github.com/google/uuid" + "github.com/mileusna/useragent" + "github.com/warmbly/warmbly/internal/config" + "github.com/warmbly/warmbly/internal/errx" + "github.com/warmbly/warmbly/internal/infrastructure/db" + "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" +) + +// Payload caps. The tracking service enforces the same limits before +// forwarding; these are the backstop for anything that reaches the backend +// another way. +const ( + maxURLLen = 2048 + maxTitleLen = 512 + maxReferrerLen = 2048 + maxLanguageLen = 32 + maxTimezoneLen = 64 + maxKeyLen = 64 + minKeyLen = 16 + maxHosts = 20 + maxHostLen = 253 + maxScreenPx = 20000 +) + +var ( + // ErrUnknownSite is a site key no workspace owns. The tracking service + // caches this negatively and budgets misses per source. + ErrUnknownSite = errors.New("unknown site key") + // ErrRejected is a well-formed hit the workspace's policy does not accept + // (tracking off, consent not met, host not allowed, automated client). + // The browser gets a quiet 204 either way. + ErrRejected = errors.New("hit rejected") + // ErrMalformed is a payload that fails validation. + ErrMalformed = errors.New("malformed hit") +) + +// identifyParam is the query parameter the click redirect appends and the +// snippet strips. Kept out of stored URLs even if a snippet forgets to. +const identifyParam = "wbly_t" + +type Service interface { + GetSettings(ctx context.Context, orgID uuid.UUID) (*models.WebsiteTrackingSettings, *errx.Error) + UpdateSettings(ctx context.Context, orgID, userID uuid.UUID, req *models.UpdateWebsiteTrackingSettingsRequest) (*models.WebsiteTrackingSettings, *errx.Error) + RotateSiteKey(ctx context.Context, orgID, userID uuid.UUID) (*models.WebsiteTrackingSettings, *errx.Error) + + // ShouldIdentify reports whether a click redirect for this campaign may + // append the ticket to the destination: only when the workspace has + // tracking on and the destination host is one it registered. + ShouldIdentify(ctx context.Context, campaignID uuid.UUID, destination string) bool + + // Ingest records one page view. Returns ErrUnknownSite, ErrRejected or + // ErrMalformed for the tracking service to map to a status. + Ingest(ctx context.Context, req *models.WebsiteHitRequest) (*models.WebsiteHitResult, error) +} + +type service struct { + repo repository.WebsiteTrackingRepository + geo *geo.Client + publisher *pubsub.StreamingPublisher +} + +func NewService(repo repository.WebsiteTrackingRepository, geoClient *geo.Client, publisher *pubsub.StreamingPublisher) Service { + return &service{repo: repo, geo: geoClient, publisher: publisher} +} + +// newKey is 128 bits of randomness as hex: unguessable, URL-safe, and the +// same shape for site keys and server-issued visitor ids. +func newKey() string { + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + return uuid.New().String() + } + return hex.EncodeToString(b) +} + +func (s *service) GetSettings(ctx context.Context, orgID uuid.UUID) (*models.WebsiteTrackingSettings, *errx.Error) { + settings, err := s.repo.GetOrCreateSettings(ctx, orgID, newKey()) + if err != nil { + db.CaptureError(err, "", nil, "websitetracking GetSettings") + return nil, errx.InternalError() + } + settings.TrackingHost = config.TrackingHost() + return settings, nil +} + +func (s *service) UpdateSettings(ctx context.Context, orgID, userID uuid.UUID, req *models.UpdateWebsiteTrackingSettingsRequest) (*models.WebsiteTrackingSettings, *errx.Error) { + current, xerr := s.GetSettings(ctx, orgID) + if xerr != nil { + return nil, xerr + } + if req.Enabled != nil { + current.Enabled = *req.Enabled + } + if req.ConsentMode != nil { + switch *req.ConsentMode { + case models.WebsiteConsentExplicit, models.WebsiteConsentImplicit: + current.ConsentMode = *req.ConsentMode + default: + return nil, errx.New(errx.BadRequest, "consent_mode must be explicit or implicit") + } + } + if req.LocationPrecision != nil { + switch *req.LocationPrecision { + case models.WebsiteLocationNone, models.WebsiteLocationCountry, models.WebsiteLocationCity: + current.LocationPrecision = *req.LocationPrecision + default: + return nil, errx.New(errx.BadRequest, "location_precision must be none, country or city") + } + } + if req.RetentionDays != nil { + if *req.RetentionDays < models.WebsiteRetentionMinDays || *req.RetentionDays > models.WebsiteRetentionMaxDays { + return nil, errx.New(errx.BadRequest, "retention_days must be between 7 and 365") + } + current.RetentionDays = *req.RetentionDays + } + if req.AllowedHosts != nil { + hosts, ok := normalizeHosts(*req.AllowedHosts) + if !ok { + return nil, errx.New(errx.BadRequest, "allowed_hosts must be up to 20 bare hostnames") + } + current.AllowedHosts = hosts + } + if err := s.repo.UpdateSettings(ctx, orgID, userID, current); err != nil { + db.CaptureError(err, "", nil, "websitetracking UpdateSettings") + return nil, errx.InternalError() + } + return s.GetSettings(ctx, orgID) +} + +func (s *service) RotateSiteKey(ctx context.Context, orgID, userID uuid.UUID) (*models.WebsiteTrackingSettings, *errx.Error) { + if _, xerr := s.GetSettings(ctx, orgID); xerr != nil { + return nil, xerr + } + if err := s.repo.RotateSiteKey(ctx, orgID, userID, newKey()); err != nil { + db.CaptureError(err, "", nil, "websitetracking RotateSiteKey") + return nil, errx.InternalError() + } + return s.GetSettings(ctx, orgID) +} + +// normalizeHosts reduces whatever was typed (URLs, mixed case, blank lines) +// to unique bare hostnames. +func normalizeHosts(raw []string) ([]string, bool) { + out := make([]string, 0, len(raw)) + seen := map[string]bool{} + for _, r := range raw { + h := config.NormalizeTrackingHost(r) + if h == "" { + continue + } + if len(h) > maxHostLen || strings.ContainsAny(h, " \t/\\") { + return nil, false + } + if seen[h] { + continue + } + seen[h] = true + out = append(out, h) + } + if len(out) > maxHosts { + return nil, false + } + return out, true +} + +// hostAllowed matches a request host against the registered list. Ports are +// ignored and a registered apex covers its subdomains, so "example.com" +// admits "www.example.com" and "app.example.com". +func hostAllowed(allowed []string, host string) bool { + host = hostOnly(host) + if host == "" { + return false + } + for _, a := range allowed { + a = hostOnly(a) + if a == "" { + continue + } + if host == a || strings.HasSuffix(host, "."+a) { + return true + } + } + return false +} + +func hostOnly(h string) string { + h = strings.ToLower(strings.TrimSpace(h)) + if hp, _, err := net.SplitHostPort(h); err == nil { + h = hp + } + return strings.TrimSuffix(h, ".") +} + +func (s *service) ShouldIdentify(ctx context.Context, campaignID uuid.UUID, destination string) bool { + site, err := s.repo.SiteForCampaign(ctx, campaignID) + if err != nil || site == nil || !site.Enabled || len(site.AllowedHosts) == 0 { + return false + } + u, perr := url.Parse(destination) + if perr != nil || (u.Scheme != "http" && u.Scheme != "https") { + return false + } + return hostAllowed(site.AllowedHosts, u.Host) +} + +func validKey(k string) bool { + if len(k) < minKeyLen || len(k) > maxKeyLen { + return false + } + for _, c := range k { + switch { + case c >= 'a' && c <= 'z', c >= 'A' && c <= 'Z', c >= '0' && c <= '9', c == '-', c == '_': + default: + return false + } + } + return true +} + +func (s *service) Ingest(ctx context.Context, req *models.WebsiteHitRequest) (*models.WebsiteHitResult, error) { + if req == nil || !validKey(req.SiteKey) || !validKey(req.VisitorKey) || !validKey(req.SessionKey) { + return nil, ErrMalformed + } + if len(req.URL) == 0 || len(req.URL) > maxURLLen || len(req.Title) > maxTitleLen || + len(req.Referrer) > maxReferrerLen || len(req.Language) > maxLanguageLen || + len(req.Timezone) > maxTimezoneLen || + req.ScreenWidth < 0 || req.ScreenHeight < 0 || req.ScreenWidth > maxScreenPx || req.ScreenHeight > maxScreenPx { + return nil, ErrMalformed + } + pageURL, err := url.Parse(req.URL) + if err != nil || (pageURL.Scheme != "http" && pageURL.Scheme != "https") || pageURL.Host == "" { + return nil, ErrMalformed + } + + site, err := s.repo.GetSiteByKey(ctx, req.SiteKey) + if err != nil { + db.CaptureError(err, "", nil, "websitetracking GetSiteByKey") + return nil, err + } + if site == nil { + return nil, ErrUnknownSite + } + if !site.Enabled { + return nil, ErrRejected + } + + // The server, not the snippet, decides whether consent was sufficient. A + // stale snippet attribute cannot downgrade an explicit-mode workspace. + switch site.ConsentMode { + case models.WebsiteConsentExplicit: + if req.Consent != "granted" { + return nil, ErrRejected + } + default: + if req.Consent != "granted" && req.Consent != "implicit" { + return nil, ErrRejected + } + } + + if len(site.AllowedHosts) > 0 { + if !hostAllowed(site.AllowedHosts, pageURL.Host) { + return nil, ErrRejected + } + if req.OriginHost != "" && !hostAllowed(site.AllowedHosts, req.OriginHost) { + return nil, ErrRejected + } + } + + ua := useragent.Parse(req.UserAgent) + if ua.Bot { + return nil, ErrRejected + } + + now := time.Now() + hit := &models.WebsitePageHit{ + SessionKey: req.SessionKey, + OccurredAt: now, + Title: strings.TrimSpace(req.Title), + Landing: req.Landing, + Language: strings.TrimSpace(req.Language), + Timezone: strings.TrimSpace(req.Timezone), + ScreenWidth: req.ScreenWidth, + ScreenHeight: req.ScreenHeight, + OS: ua.OS, + Browser: ua.Name, + BrowserVersion: ua.Version, + DeviceBrand: ua.Device, + DeviceType: deviceType(ua), + } + + // Strip the identification ticket from the stored URL so it never shows + // up in the timeline or travels in an archive. + q := pageURL.Query() + hit.UTMSource = clip(q.Get("utm_source"), 256) + hit.UTMMedium = clip(q.Get("utm_medium"), 256) + hit.UTMCampaign = clip(q.Get("utm_campaign"), 256) + hit.UTMTerm = clip(q.Get("utm_term"), 256) + hit.UTMContent = clip(q.Get("utm_content"), 256) + if q.Has(identifyParam) { + q.Del(identifyParam) + pageURL.RawQuery = q.Encode() + } + pageURL.Fragment = "" + hit.URL = pageURL.String() + hit.Path = pageURL.Path + if hit.Path == "" { + hit.Path = "/" + } + + if ref := strings.TrimSpace(req.Referrer); ref != "" { + if ru, rerr := url.Parse(ref); rerr == nil && (ru.Scheme == "http" || ru.Scheme == "https") { + ru.Fragment = "" + hit.Referrer = ru.String() + hit.ReferrerDomain = hostOnly(ru.Host) + } + } + + s.locate(req.IP, site.LocationPrecision, hit) + + visitor, err := s.repo.UpsertVisitor(ctx, site.OrganizationID, req.VisitorKey, now) + if err != nil { + db.CaptureError(err, "", nil, "websitetracking UpsertVisitor") + return nil, err + } + + result := &models.WebsiteHitResult{} + if req.IdentifyToken != "" { + if ticket, perr := uuid.Parse(req.IdentifyToken); perr == nil { + contactID, ticketOrg, ok, terr := s.repo.ContactForTicket(ctx, ticket) + if terr != nil { + db.CaptureError(terr, "", nil, "websitetracking ContactForTicket") + } + // A ticket from another workspace's campaign proves nothing here. + if terr == nil && ok && ticketOrg == site.OrganizationID { + switch { + case visitor.ContactID == nil: + if ierr := s.repo.IdentifyVisitor(ctx, visitor.ID, contactID, "email_link", now); ierr != nil { + db.CaptureError(ierr, "", nil, "websitetracking IdentifyVisitor") + } else { + visitor.ContactID = &contactID + } + case *visitor.ContactID != contactID: + // A different person on the same browser: split rather than + // merge, so neither contact inherits the other's history. + fresh, cerr := s.repo.CreateVisitor(ctx, site.OrganizationID, newKey(), contactID, "email_link", now) + if cerr != nil { + db.CaptureError(cerr, "", nil, "websitetracking CreateVisitor") + } else { + visitor = fresh + result.NewVisitorKey = fresh.VisitorKey + } + } + } + } + } + + hit.VisitorID = visitor.ID + if err := s.repo.InsertHit(ctx, site.OrganizationID, hit); err != nil { + db.CaptureError(err, "", nil, "websitetracking InsertHit") + return nil, err + } + + if visitor.ContactID != nil && s.publisher != nil { + s.publisher.PublishPageHit(ctx, &pubsub.PageHitEvent{ + OrgID: site.OrganizationID.String(), + ContactID: visitor.ContactID.String(), + URL: hit.URL, + Title: hit.Title, + }) + } + return result, nil +} + +// locate fills the location columns from the request IP, trimmed to the +// workspace's precision. Best-effort: no database or a private address leaves +// them empty. The IP itself goes no further than this call. +func (s *service) locate(ip string, precision models.WebsiteLocationPrecision, hit *models.WebsitePageHit) { + if precision == models.WebsiteLocationNone || s.geo == nil || ip == "" { + return + } + addr, err := netip.ParseAddr(strings.TrimSpace(ip)) + if err != nil || addr.IsPrivate() || addr.IsLoopback() { + return + } + info, err := s.geo.Lookup(addr) + if err != nil || info == nil { + return + } + hit.CountryCode = info.CountryCode + if precision == models.WebsiteLocationCity { + hit.Region = info.Region + if info.City != "Unknown" { + hit.City = info.City + } + } +} + +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" + } +} + +func clip(s string, n int) string { + s = strings.TrimSpace(s) + if len(s) > n { + return s[:n] + } + return s +} diff --git a/internal/app/websitetracking/service_test.go b/internal/app/websitetracking/service_test.go new file mode 100644 index 00000000..ca8bbbe6 --- /dev/null +++ b/internal/app/websitetracking/service_test.go @@ -0,0 +1,53 @@ +package websitetracking + +import "testing" + +func TestHostAllowedCoversSubdomainsAndIgnoresPorts(t *testing.T) { + allowed := []string{"example.com", "shop.other.io"} + for host, want := range map[string]bool{ + "example.com": true, + "www.example.com": true, + "app.example.com:8443": true, + "EXAMPLE.COM": true, + "notexample.com": false, + "example.com.evil.net": false, + "other.io": false, + "shop.other.io": true, + "": false, + } { + if got := hostAllowed(allowed, host); got != want { + t.Errorf("hostAllowed(%q) = %v, want %v", host, got, want) + } + } +} + +func TestNormalizeHostsReducesPastedURLs(t *testing.T) { + hosts, ok := normalizeHosts([]string{" https://WWW.Example.com/path ", "www.example.com", "", "app.example.com."}) + if !ok { + t.Fatal("expected hosts to normalize") + } + if len(hosts) != 2 || hosts[0] != "www.example.com" || hosts[1] != "app.example.com" { + t.Fatalf("unexpected hosts: %v", hosts) + } + if _, ok := normalizeHosts(make([]string, 0)); !ok { + t.Fatal("empty list must be allowed") + } + many := make([]string, maxHosts+1) + for i := range many { + many[i] = "h" + string(rune('a'+i)) + ".example.com" + } + if _, ok := normalizeHosts(many); ok { + t.Fatal("expected the host cap to refuse") + } +} + +func TestValidKeyShape(t *testing.T) { + if !validKey("0123456789abcdef0123456789abcdef") { + t.Fatal("hex key must pass") + } + for _, bad := range []string{"short", "has space in the key!!", ""} { + if validKey(bad) { + t.Errorf("validKey(%q) should fail", bad) + } + } +} diff --git a/internal/infrastructure/db/migrations/000106_website_tracking.down.sql b/internal/infrastructure/db/migrations/000106_website_tracking.down.sql new file mode 100644 index 00000000..a0d0fe7e --- /dev/null +++ b/internal/infrastructure/db/migrations/000106_website_tracking.down.sql @@ -0,0 +1,3 @@ +DROP TABLE IF EXISTS public.website_page_hits; +DROP TABLE IF EXISTS public.website_visitors; +DROP TABLE IF EXISTS public.website_tracking_settings; diff --git a/internal/infrastructure/db/migrations/000106_website_tracking.up.sql b/internal/infrastructure/db/migrations/000106_website_tracking.up.sql new file mode 100644 index 00000000..661e1156 --- /dev/null +++ b/internal/infrastructure/db/migrations/000106_website_tracking.up.sql @@ -0,0 +1,115 @@ +-- Website visitor tracking (issue #255, section 13). +-- +-- A workspace can install a snippet on its own site and see which pages a +-- contact visited in the contact timeline. Three relations: +-- +-- website_tracking_settings one row per workspace: the public site key the +-- snippet carries, the consent mode, the location +-- precision and the retention window +-- website_visitors one row per browser (the visitor id the snippet +-- keeps in first-party storage), linked to a +-- contact once an email-link ticket identifies it +-- website_page_hits one row per counted page view, enriched on the +-- server (device from the user agent, location +-- from the IP). The IP itself is never stored. +-- +-- Retention is enforced by the backend job in internal/jobs; the window is +-- per workspace and bounded by the CHECK below. + +CREATE TABLE public.website_tracking_settings ( + organization_id uuid PRIMARY KEY REFERENCES public.organizations(id) ON DELETE CASCADE, + enabled boolean NOT NULL DEFAULT false, + -- Public identifier embedded in the snippet. Not a secret: it only routes a + -- hit to a workspace and can be rotated from the dashboard. + site_key text NOT NULL UNIQUE, + -- explicit: nothing is recorded until the page calls warmbly('consent','granted'). + -- implicit: recorded on load; the workspace asserts its own lawful basis. + consent_mode text NOT NULL DEFAULT 'explicit' + CHECK (consent_mode IN ('explicit', 'implicit')), + -- How much of the IP-derived location is kept: none | country | city. + location_precision text NOT NULL DEFAULT 'country' + CHECK (location_precision IN ('none', 'country', 'city')), + -- Hosts the snippet may report from, and the only hosts a click redirect + -- appends the identification ticket to. Empty means any host. + allowed_hosts text[] NOT NULL DEFAULT '{}', + retention_days integer NOT NULL DEFAULT 90 + CHECK (retention_days BETWEEN 7 AND 365), + updated_by uuid REFERENCES public.users(id) ON DELETE SET NULL, + created_at timestamptz NOT NULL DEFAULT now(), + updated_at timestamptz NOT NULL DEFAULT now() +); + +COMMENT ON TABLE public.website_tracking_settings IS + 'Per-workspace website tracking configuration: snippet site key, consent mode, location precision, retention.'; + +CREATE TABLE public.website_visitors ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + organization_id uuid NOT NULL REFERENCES public.organizations(id) ON DELETE CASCADE, + -- The id the snippet keeps in the browser's first-party storage. + visitor_key text NOT NULL, + -- Deleting a contact erases their browsing history with them. + contact_id uuid REFERENCES public.contacts(id) ON DELETE CASCADE, + identified_at timestamptz, + -- How the browser was tied to the contact: email_link (a click ticket). + identified_via text, + first_seen_at timestamptz NOT NULL DEFAULT now(), + last_seen_at timestamptz NOT NULL DEFAULT now(), + UNIQUE (organization_id, visitor_key) +); + +CREATE INDEX idx_website_visitors_contact + ON public.website_visitors USING btree (contact_id) + WHERE contact_id IS NOT NULL; + +CREATE INDEX idx_website_visitors_last_seen + ON public.website_visitors USING btree (organization_id, last_seen_at); + +COMMENT ON TABLE public.website_visitors IS + 'One row per browser the tracking snippet has seen, linked to a contact once an email-link ticket identifies it.'; + +CREATE TABLE public.website_page_hits ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + organization_id uuid NOT NULL REFERENCES public.organizations(id) ON DELETE CASCADE, + visitor_id uuid NOT NULL REFERENCES public.website_visitors(id) ON DELETE CASCADE, + session_key text NOT NULL, + occurred_at timestamptz NOT NULL DEFAULT now(), + url text NOT NULL, + path text NOT NULL DEFAULT '', + title text NOT NULL DEFAULT '', + referrer text NOT NULL DEFAULT '', + referrer_domain text NOT NULL DEFAULT '', + -- First hit of a session: the landing page. + landing boolean NOT NULL DEFAULT false, + utm_source text NOT NULL DEFAULT '', + utm_medium text NOT NULL DEFAULT '', + utm_campaign text NOT NULL DEFAULT '', + utm_term text NOT NULL DEFAULT '', + utm_content text NOT NULL DEFAULT '', + -- Parsed on the server from the User-Agent header, never from the body. + device_type text NOT NULL DEFAULT 'unknown' + CHECK (device_type IN ('desktop', 'mobile', 'tablet', 'unknown')), + os text NOT NULL DEFAULT '', + browser text NOT NULL DEFAULT '', + browser_version text NOT NULL DEFAULT '', + device_brand text NOT NULL DEFAULT '', + -- Reported by the browser: cheap, harmless, and not derivable server-side. + language text NOT NULL DEFAULT '', + timezone text NOT NULL DEFAULT '', + screen_width integer NOT NULL DEFAULT 0, + screen_height integer NOT NULL DEFAULT 0, + -- Resolved on the server from the request IP, trimmed to the workspace's + -- location precision. The IP is not stored. + country_code text NOT NULL DEFAULT '', + region text NOT NULL DEFAULT '', + city text NOT NULL DEFAULT '' +); + +CREATE INDEX idx_website_page_hits_visitor_recent + ON public.website_page_hits USING btree (visitor_id, occurred_at DESC); + +-- The retention sweep deletes by workspace and age. +CREATE INDEX idx_website_page_hits_org_age + ON public.website_page_hits USING btree (organization_id, occurred_at); + +COMMENT ON TABLE public.website_page_hits IS + 'Counted page views from the tracking snippet, enriched server-side. Pruned per workspace by the retention job.'; diff --git a/internal/infrastructure/pubsub/events.go b/internal/infrastructure/pubsub/events.go index a97ab4d8..6173865e 100644 --- a/internal/infrastructure/pubsub/events.go +++ b/internal/infrastructure/pubsub/events.go @@ -60,6 +60,10 @@ const ( // A human reply landed for a campaign contact (org-scoped pulse). EventEmailReplied EventType = "EMAIL_REPLIED" + // A website page view landed for an identified contact (org-scoped; the + // dashboard refreshes that contact's timeline). + EventPageHit EventType = "PAGE_HIT" + // Task progress events EventTaskProgress EventType = "TASK_PROGRESS" @@ -212,6 +216,31 @@ type TrackingEventPayload struct { Machine bool `json:"machine,omitempty"` } +// PageHitEvent is a website page view tied to a contact. +type PageHitEvent struct { + BaseEvent + OrgID string `json:"org_id"` + ContactID string `json:"contact_id"` + URL string `json:"url"` + Title string `json:"title,omitempty"` +} + +// PublishPageHit tells the org that a contact viewed a page. No user id: the +// event belongs to the workspace, not to any member. +func (p *StreamingPublisher) PublishPageHit(ctx context.Context, event *PageHitEvent) { + if p.client == nil { + return + } + event.EventType = EventPageHit + event.Timestamp = time.Now() + attrs := map[string]string{ + "org_id": event.OrgID, + "contact_id": event.ContactID, + "event_type": string(event.EventType), + } + _ = p.client.Publish(ctx, TopicCampaignUpdate, event, attrs) +} + // TaskProgressEvent for detailed campaign task progress type TaskProgressEvent struct { BaseEvent diff --git a/internal/jobs/website_tracking_retention.go b/internal/jobs/website_tracking_retention.go new file mode 100644 index 00000000..1d0247ab --- /dev/null +++ b/internal/jobs/website_tracking_retention.go @@ -0,0 +1,56 @@ +package jobs + +import ( + "context" + "time" + + "github.com/getsentry/sentry-go" + "github.com/warmbly/warmbly/internal/repository" +) + +// WebsiteTrackingRetentionJob prunes page views past each workspace's own +// retention window. The window is the promise the workspace makes on its +// tracking settings page, so the sweep is what keeps that promise. +type WebsiteTrackingRetentionJob struct { + repo repository.WebsiteTrackingRepository +} + +func NewWebsiteTrackingRetentionJob(repo repository.WebsiteTrackingRepository) *WebsiteTrackingRetentionJob { + return &WebsiteTrackingRetentionJob{repo: repo} +} + +// Run executes one pruning pass over every workspace. A failure on one +// workspace is reported and the pass continues with the next. +func (j *WebsiteTrackingRetentionJob) Run(ctx context.Context) error { + if j.repo == nil { + return nil + } + cutoffs, err := j.repo.RetentionCutoffs(ctx, time.Now()) + if err != nil { + sentry.CaptureException(err) + return err + } + var last error + for _, c := range cutoffs { + if _, err := j.repo.PruneBefore(ctx, c.OrganizationID, c.Before); err != nil { + sentry.CaptureException(err) + last = err + } + } + return last +} + +// Start runs the job once on boot and then on the interval until ctx ends. +func (j *WebsiteTrackingRetentionJob) Start(ctx context.Context, interval time.Duration) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + _ = j.Run(ctx) + for { + select { + case <-ticker.C: + _ = j.Run(ctx) + case <-ctx.Done(): + return + } + } +} diff --git a/internal/models/contact.go b/internal/models/contact.go index 59c3baa4..6de54357 100644 --- a/internal/models/contact.go +++ b/internal/models/contact.go @@ -250,6 +250,10 @@ const ( TimelineMeetingBooked ContactTimelineEventType = "meeting_booked" TimelineMeetingRescheduled ContactTimelineEventType = "meeting_rescheduled" TimelineMeetingCanceled ContactTimelineEventType = "meeting_canceled" + + // A page view reported by the website tracking snippet, tied to the + // contact through an email-link ticket. Detail rides in PageHit. + TimelinePageHit ContactTimelineEventType = "page_hit" ) // ContactTimelineEvent is one entry in the merged activity feed. The @@ -289,6 +293,9 @@ type ContactTimelineEvent struct { // Author (notes). UserID *uuid.UUID `json:"user_id,omitempty"` + + // Website page view (page_hit): URL, referrer, device, UTM, location. + PageHit *WebsitePageHit `json:"page_hit,omitempty"` } type ContactTimelineResult struct { diff --git a/internal/models/website_tracking.go b/internal/models/website_tracking.go new file mode 100644 index 00000000..493d9ba1 --- /dev/null +++ b/internal/models/website_tracking.go @@ -0,0 +1,161 @@ +package models + +import ( + "time" + + "github.com/google/uuid" +) + +// Website visitor tracking (issue #255, section 13): the snippet a workspace +// installs on its own site, and the page views it reports. + +// WebsiteConsentMode says when the snippet may record anything. +type WebsiteConsentMode string + +const ( + // WebsiteConsentExplicit records nothing until the page calls + // warmbly('consent', 'granted'). The default, because the product sells into + // jurisdictions where device and location data need a prior opt-in. + WebsiteConsentExplicit WebsiteConsentMode = "explicit" + // WebsiteConsentImplicit records on load. The workspace asserts its own + // lawful basis by choosing it. + WebsiteConsentImplicit WebsiteConsentMode = "implicit" +) + +// WebsiteLocationPrecision is how much of the IP-derived location is kept. +type WebsiteLocationPrecision string + +const ( + WebsiteLocationNone WebsiteLocationPrecision = "none" + WebsiteLocationCountry WebsiteLocationPrecision = "country" + WebsiteLocationCity WebsiteLocationPrecision = "city" +) + +const ( + WebsiteRetentionMinDays = 7 + WebsiteRetentionMaxDays = 365 + WebsiteRetentionDefaultDays = 90 +) + +// WebsiteTrackingSettings is a workspace's tracking configuration. Created on +// first read with tracking disabled, so every workspace has a site key to show +// but nothing is accepted until someone turns it on. +type WebsiteTrackingSettings struct { + OrganizationID uuid.UUID `json:"organization_id"` + Enabled bool `json:"enabled"` + SiteKey string `json:"site_key"` + ConsentMode WebsiteConsentMode `json:"consent_mode"` + LocationPrecision WebsiteLocationPrecision `json:"location_precision"` + AllowedHosts []string `json:"allowed_hosts"` + RetentionDays int `json:"retention_days"` + UpdatedAt time.Time `json:"updated_at"` + // TrackingHost is the deployment's tracking host (TRACKING_DOMAIN), so the + // dashboard can render the exact snippet. Empty when the install has none. + TrackingHost string `json:"tracking_host"` +} + +// UpdateWebsiteTrackingSettingsRequest is the PATCH body. Every field is +// optional; absent fields keep their value. +type UpdateWebsiteTrackingSettingsRequest struct { + Enabled *bool `json:"enabled"` + ConsentMode *WebsiteConsentMode `json:"consent_mode"` + LocationPrecision *WebsiteLocationPrecision `json:"location_precision"` + AllowedHosts *[]string `json:"allowed_hosts"` + RetentionDays *int `json:"retention_days"` +} + +// WebsiteHitRequest is what the tracking service forwards to the backend for +// one page view: the snippet's payload plus the request facts only the edge +// saw. Device and location are derived here from UserAgent and IP; nothing +// about them is trusted from the snippet. +type WebsiteHitRequest struct { + SiteKey string `json:"site_key"` + VisitorKey string `json:"visitor_key"` + SessionKey string `json:"session_key"` + // Consent is what the snippet believes: "granted" after an explicit + // opt-in, "implicit" when the snippet runs in implicit mode. The + // workspace's configured mode decides whether that is enough. + Consent string `json:"consent"` + // IdentifyToken is the click ticket the redirect appended to the landing + // URL. It is the only way a hit reaches a contact. + IdentifyToken string `json:"identify_token"` + + URL string `json:"url"` + Title string `json:"title"` + Referrer string `json:"referrer"` + Language string `json:"language"` + Timezone string `json:"timezone"` + ScreenWidth int `json:"screen_width"` + ScreenHeight int `json:"screen_height"` + // Landing marks the first view of a session, as judged by the snippet. + Landing bool `json:"landing"` + + UserAgent string `json:"user_agent"` + IP string `json:"ip"` + OriginHost string `json:"origin_host"` +} + +// WebsiteHitResult tells the tracking service what to answer the browser. +type WebsiteHitResult struct { + // NewVisitorKey is set when the browser must adopt a fresh visitor id: + // the ticket named a different contact than the one already tied to + // this browser, so the record was split rather than merged. + NewVisitorKey string `json:"new_visitor_key,omitempty"` +} + +// WebsitePageHit is one counted page view as stored and as shown in the +// contact timeline. +type WebsitePageHit struct { + ID uuid.UUID `json:"id"` + VisitorID uuid.UUID `json:"visitor_id"` + SessionKey string `json:"session_key"` + OccurredAt time.Time `json:"occurred_at"` + URL string `json:"url"` + Path string `json:"path"` + Title string `json:"title"` + Referrer string `json:"referrer"` + ReferrerDomain string `json:"referrer_domain"` + Landing bool `json:"landing"` + UTMSource string `json:"utm_source"` + UTMMedium string `json:"utm_medium"` + UTMCampaign string `json:"utm_campaign"` + UTMTerm string `json:"utm_term"` + UTMContent string `json:"utm_content"` + DeviceType string `json:"device_type"` + OS string `json:"os"` + Browser string `json:"browser"` + BrowserVersion string `json:"browser_version"` + DeviceBrand string `json:"device_brand"` + Language string `json:"language"` + Timezone string `json:"timezone"` + ScreenWidth int `json:"screen_width"` + ScreenHeight int `json:"screen_height"` + CountryCode string `json:"country_code"` + Region string `json:"region"` + City string `json:"city"` +} + +// WebsiteSite is what the ingest path needs to know about a site key. +type WebsiteSite struct { + OrganizationID uuid.UUID + Enabled bool + ConsentMode WebsiteConsentMode + LocationPrecision WebsiteLocationPrecision + AllowedHosts []string +} + +// WebsiteVisitor is one browser the snippet has seen. +type WebsiteVisitor struct { + ID uuid.UUID + OrganizationID uuid.UUID + VisitorKey string + ContactID *uuid.UUID + IdentifiedAt *time.Time + IdentifiedVia string +} + +// WebsiteTrackingRetentionCutoff is one workspace's prune boundary. +type WebsiteTrackingRetentionCutoff struct { + OrganizationID uuid.UUID + Before time.Time +} diff --git a/internal/repository/pg_contact.go b/internal/repository/pg_contact.go index 89cc5834..3015c7ca 100644 --- a/internal/repository/pg_contact.go +++ b/internal/repository/pg_contact.go @@ -2419,6 +2419,7 @@ func (r *contactRepository) ListSentEmails(ctx context.Context, userID, contactI // - deliverability_events → bounce / complaint // - suppressed_recipients → suppression added // - contact_notes → CRM notes +// - website_page_hits → page views from the tracking snippet // // We pull up to (limit) candidates from each source ordered by time // DESC, then merge-sort in Go. This avoids a 5-way UNION with @@ -2717,6 +2718,59 @@ func (r *contactRepository) ListTimeline(ctx context.Context, userID uuid.UUID, events = append(events, ev) } mrows.Close() + + // 7. Website page views from any browser tied to the contact through + // an email-link ticket. The detail block carries everything the + // expandable row shows. + hitQuery := ` + SELECT h.id, h.visitor_id, h.session_key, h.occurred_at, + h.url, h.path, h.title, h.referrer, h.referrer_domain, h.landing, + h.utm_source, h.utm_medium, h.utm_campaign, h.utm_term, h.utm_content, + h.device_type, h.os, h.browser, h.browser_version, h.device_brand, + h.language, h.timezone, h.screen_width, h.screen_height, + h.country_code, h.region, h.city + FROM website_page_hits h + WHERE h.organization_id = $1 + AND h.visitor_id IN (SELECT id FROM website_visitors WHERE contact_id = $2) + AND h.occurred_at < $3 + ORDER BY h.occurred_at DESC + LIMIT $4 + ` + hrows, err := r.DB.Query(ctx, hitQuery, *orgID, contactID, bound, limit) + if err != nil { + db.CaptureError(err, hitQuery, nil, "ListTimeline page hits") + return nil, errx.InternalError() + } + for hrows.Next() { + var h models.WebsitePageHit + if err := hrows.Scan( + &h.ID, &h.VisitorID, &h.SessionKey, &h.OccurredAt, + &h.URL, &h.Path, &h.Title, &h.Referrer, &h.ReferrerDomain, &h.Landing, + &h.UTMSource, &h.UTMMedium, &h.UTMCampaign, &h.UTMTerm, &h.UTMContent, + &h.DeviceType, &h.OS, &h.Browser, &h.BrowserVersion, &h.DeviceBrand, + &h.Language, &h.Timezone, &h.ScreenWidth, &h.ScreenHeight, + &h.CountryCode, &h.Region, &h.City, + ); err != nil { + hrows.Close() + db.CaptureError(err, "", nil, "ListTimeline page hits scan") + return nil, errx.InternalError() + } + hit := h + ev := models.ContactTimelineEvent{ + Type: models.TimelinePageHit, + At: h.OccurredAt, + PageHit: &hit, + } + if h.Title != "" { + title := h.Title + ev.Subject = &title + } else { + path := h.Path + ev.Subject = &path + } + events = append(events, ev) + } + hrows.Close() } // Merge sort: newest first. diff --git a/internal/repository/pg_website_tracking.go b/internal/repository/pg_website_tracking.go new file mode 100644 index 00000000..a6d84ce6 --- /dev/null +++ b/internal/repository/pg_website_tracking.go @@ -0,0 +1,302 @@ +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" +) + +// WebsiteTrackingRepository is the store behind the website tracking snippet: +// per-workspace settings, the browsers the snippet has seen, and their page +// views. Writes on the hit path come only from the backend's internal ingest +// endpoint; the dashboard only reads hits. +type WebsiteTrackingRepository interface { + GetOrCreateSettings(ctx context.Context, orgID uuid.UUID, siteKey string) (*models.WebsiteTrackingSettings, error) + UpdateSettings(ctx context.Context, orgID, updatedBy uuid.UUID, s *models.WebsiteTrackingSettings) error + RotateSiteKey(ctx context.Context, orgID, updatedBy uuid.UUID, siteKey string) error + + GetSiteByKey(ctx context.Context, siteKey string) (*models.WebsiteSite, error) + SiteForCampaign(ctx context.Context, campaignID uuid.UUID) (*models.WebsiteSite, error) + // ContactForTicket resolves a click ticket to the contact it was mailed to + // and the workspace that owns the campaign. ok is false for an unknown + // ticket or one whose task has no contact. + ContactForTicket(ctx context.Context, ticketID uuid.UUID) (contactID, orgID uuid.UUID, ok bool, err error) + + UpsertVisitor(ctx context.Context, orgID uuid.UUID, visitorKey string, seenAt time.Time) (*models.WebsiteVisitor, error) + CreateVisitor(ctx context.Context, orgID uuid.UUID, visitorKey string, contactID uuid.UUID, via string, at time.Time) (*models.WebsiteVisitor, error) + IdentifyVisitor(ctx context.Context, visitorID, contactID uuid.UUID, via string, at time.Time) error + InsertHit(ctx context.Context, orgID uuid.UUID, hit *models.WebsitePageHit) error + + ListHitsForContact(ctx context.Context, orgID, contactID uuid.UUID, before time.Time, limit int) ([]models.WebsitePageHit, error) + + RetentionCutoffs(ctx context.Context, now time.Time) ([]models.WebsiteTrackingRetentionCutoff, error) + PruneBefore(ctx context.Context, orgID uuid.UUID, before time.Time) (int64, error) +} + +type websiteTrackingRepository struct { + db *pgxpool.Pool +} + +func NewWebsiteTrackingRepository(db *pgxpool.Pool) WebsiteTrackingRepository { + return &websiteTrackingRepository{db: db} +} + +const websiteSettingsColumns = `organization_id, enabled, site_key, consent_mode, location_precision, allowed_hosts, retention_days, updated_at` + +func scanWebsiteSettings(row pgx.Row) (*models.WebsiteTrackingSettings, error) { + var s models.WebsiteTrackingSettings + if err := row.Scan(&s.OrganizationID, &s.Enabled, &s.SiteKey, &s.ConsentMode, &s.LocationPrecision, &s.AllowedHosts, &s.RetentionDays, &s.UpdatedAt); err != nil { + return nil, err + } + if s.AllowedHosts == nil { + s.AllowedHosts = []string{} + } + return &s, nil +} + +// GetOrCreateSettings returns the workspace's row, creating a disabled one +// with the given site key on first read so the dashboard always has a key to +// show. +func (r *websiteTrackingRepository) GetOrCreateSettings(ctx context.Context, orgID uuid.UUID, siteKey string) (*models.WebsiteTrackingSettings, error) { + _, err := r.db.Exec(ctx, ` + INSERT INTO website_tracking_settings (organization_id, site_key) + VALUES ($1, $2) + ON CONFLICT (organization_id) DO NOTHING + `, orgID, siteKey) + if err != nil { + return nil, err + } + return scanWebsiteSettings(r.db.QueryRow(ctx, + `SELECT `+websiteSettingsColumns+` FROM website_tracking_settings WHERE organization_id = $1`, orgID)) +} + +func (r *websiteTrackingRepository) UpdateSettings(ctx context.Context, orgID, updatedBy uuid.UUID, s *models.WebsiteTrackingSettings) error { + _, err := r.db.Exec(ctx, ` + UPDATE website_tracking_settings + SET enabled = $2, consent_mode = $3, location_precision = $4, allowed_hosts = $5, + retention_days = $6, updated_by = $7, updated_at = now() + WHERE organization_id = $1 + `, orgID, s.Enabled, s.ConsentMode, s.LocationPrecision, s.AllowedHosts, s.RetentionDays, updatedBy) + return err +} + +func (r *websiteTrackingRepository) RotateSiteKey(ctx context.Context, orgID, updatedBy uuid.UUID, siteKey string) error { + _, err := r.db.Exec(ctx, ` + UPDATE website_tracking_settings + SET site_key = $2, updated_by = $3, updated_at = now() + WHERE organization_id = $1 + `, orgID, siteKey, updatedBy) + return err +} + +func scanWebsiteSite(row pgx.Row) (*models.WebsiteSite, error) { + var s models.WebsiteSite + err := row.Scan(&s.OrganizationID, &s.Enabled, &s.ConsentMode, &s.LocationPrecision, &s.AllowedHosts) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + return &s, nil +} + +// GetSiteByKey is the ingest lookup. nil, nil when the key is unknown. +func (r *websiteTrackingRepository) GetSiteByKey(ctx context.Context, siteKey string) (*models.WebsiteSite, error) { + return scanWebsiteSite(r.db.QueryRow(ctx, ` + SELECT organization_id, enabled, consent_mode, location_precision, allowed_hosts + FROM website_tracking_settings + WHERE site_key = $1 + `, siteKey)) +} + +// SiteForCampaign is what the click redirect consults to decide whether to +// append the identification ticket. nil, nil when the workspace never opened +// the tracking settings. +func (r *websiteTrackingRepository) SiteForCampaign(ctx context.Context, campaignID uuid.UUID) (*models.WebsiteSite, error) { + return scanWebsiteSite(r.db.QueryRow(ctx, ` + SELECT s.organization_id, s.enabled, s.consent_mode, s.location_precision, s.allowed_hosts + FROM campaigns c + JOIN website_tracking_settings s ON s.organization_id = c.organization_id + WHERE c.id = $1 + `, campaignID)) +} + +func (r *websiteTrackingRepository) ContactForTicket(ctx context.Context, ticketID uuid.UUID) (uuid.UUID, uuid.UUID, bool, error) { + var contactID, orgID *uuid.UUID + err := r.db.QueryRow(ctx, ` + SELECT ct.contact_id, c.organization_id + FROM tracked_links tl + JOIN campaign_tasks ct ON ct.task_id = tl.task_id + JOIN campaigns c ON c.id = tl.campaign_id + WHERE tl.id = $1 + `, ticketID).Scan(&contactID, &orgID) + if errors.Is(err, pgx.ErrNoRows) { + return uuid.Nil, uuid.Nil, false, nil + } + if err != nil { + return uuid.Nil, uuid.Nil, false, err + } + if contactID == nil || orgID == nil { + return uuid.Nil, uuid.Nil, false, nil + } + return *contactID, *orgID, true, nil +} + +const websiteVisitorColumns = `id, organization_id, visitor_key, contact_id, identified_at, COALESCE(identified_via, '')` + +func scanWebsiteVisitor(row pgx.Row) (*models.WebsiteVisitor, error) { + var v models.WebsiteVisitor + if err := row.Scan(&v.ID, &v.OrganizationID, &v.VisitorKey, &v.ContactID, &v.IdentifiedAt, &v.IdentifiedVia); err != nil { + return nil, err + } + return &v, nil +} + +// UpsertVisitor returns the browser's row, creating it on first sight and +// bumping last_seen_at otherwise. +func (r *websiteTrackingRepository) UpsertVisitor(ctx context.Context, orgID uuid.UUID, visitorKey string, seenAt time.Time) (*models.WebsiteVisitor, error) { + return scanWebsiteVisitor(r.db.QueryRow(ctx, ` + INSERT INTO website_visitors (organization_id, visitor_key, first_seen_at, last_seen_at) + VALUES ($1, $2, $3, $3) + ON CONFLICT (organization_id, visitor_key) + DO UPDATE SET last_seen_at = GREATEST(website_visitors.last_seen_at, EXCLUDED.last_seen_at) + RETURNING `+websiteVisitorColumns, orgID, visitorKey, seenAt)) +} + +// CreateVisitor opens a fresh, already-identified browser record. Used when a +// ticket names a different contact than the one tied to the browser: the old +// record keeps its history and the browser moves to this one. +func (r *websiteTrackingRepository) CreateVisitor(ctx context.Context, orgID uuid.UUID, visitorKey string, contactID uuid.UUID, via string, at time.Time) (*models.WebsiteVisitor, error) { + return scanWebsiteVisitor(r.db.QueryRow(ctx, ` + INSERT INTO website_visitors (organization_id, visitor_key, contact_id, identified_at, identified_via, first_seen_at, last_seen_at) + VALUES ($1, $2, $3, $4, $5, $4, $4) + RETURNING `+websiteVisitorColumns, orgID, visitorKey, contactID, at, via)) +} + +// IdentifyVisitor ties an anonymous browser to a contact. A browser already +// tied to someone is left alone; the caller splits it instead. +func (r *websiteTrackingRepository) IdentifyVisitor(ctx context.Context, visitorID, contactID uuid.UUID, via string, at time.Time) error { + _, err := r.db.Exec(ctx, ` + UPDATE website_visitors + SET contact_id = $2, identified_at = $3, identified_via = $4 + WHERE id = $1 AND contact_id IS NULL + `, visitorID, contactID, at, via) + return err +} + +func (r *websiteTrackingRepository) InsertHit(ctx context.Context, orgID uuid.UUID, h *models.WebsitePageHit) error { + if h.ID == uuid.Nil { + h.ID = uuid.New() + } + _, err := r.db.Exec(ctx, ` + INSERT INTO website_page_hits ( + id, organization_id, visitor_id, session_key, occurred_at, + url, path, title, referrer, referrer_domain, landing, + utm_source, utm_medium, utm_campaign, utm_term, utm_content, + device_type, os, browser, browser_version, device_brand, + language, timezone, screen_width, screen_height, + country_code, region, city + ) VALUES ( + $1, $2, $3, $4, $5, + $6, $7, $8, $9, $10, $11, + $12, $13, $14, $15, $16, + $17, $18, $19, $20, $21, + $22, $23, $24, $25, + $26, $27, $28 + ) + `, + h.ID, orgID, h.VisitorID, h.SessionKey, h.OccurredAt, + h.URL, h.Path, h.Title, h.Referrer, h.ReferrerDomain, h.Landing, + h.UTMSource, h.UTMMedium, h.UTMCampaign, h.UTMTerm, h.UTMContent, + h.DeviceType, h.OS, h.Browser, h.BrowserVersion, h.DeviceBrand, + h.Language, h.Timezone, h.ScreenWidth, h.ScreenHeight, + h.CountryCode, h.Region, h.City, + ) + return err +} + +// ListHitsForContact feeds the contact timeline: every counted view from any +// browser tied to the contact, newest first, strictly before the cursor. +func (r *websiteTrackingRepository) ListHitsForContact(ctx context.Context, orgID, contactID uuid.UUID, before time.Time, limit int) ([]models.WebsitePageHit, error) { + rows, err := r.db.Query(ctx, ` + SELECT h.id, h.visitor_id, h.session_key, h.occurred_at, + h.url, h.path, h.title, h.referrer, h.referrer_domain, h.landing, + h.utm_source, h.utm_medium, h.utm_campaign, h.utm_term, h.utm_content, + h.device_type, h.os, h.browser, h.browser_version, h.device_brand, + h.language, h.timezone, h.screen_width, h.screen_height, + h.country_code, h.region, h.city + FROM website_page_hits h + WHERE h.organization_id = $1 + AND h.visitor_id IN (SELECT id FROM website_visitors WHERE contact_id = $2) + AND h.occurred_at < $3 + ORDER BY h.occurred_at DESC + LIMIT $4 + `, orgID, contactID, before, limit) + if err != nil { + return nil, err + } + defer rows.Close() + + out := make([]models.WebsitePageHit, 0, limit) + for rows.Next() { + var h models.WebsitePageHit + if err := rows.Scan( + &h.ID, &h.VisitorID, &h.SessionKey, &h.OccurredAt, + &h.URL, &h.Path, &h.Title, &h.Referrer, &h.ReferrerDomain, &h.Landing, + &h.UTMSource, &h.UTMMedium, &h.UTMCampaign, &h.UTMTerm, &h.UTMContent, + &h.DeviceType, &h.OS, &h.Browser, &h.BrowserVersion, &h.DeviceBrand, + &h.Language, &h.Timezone, &h.ScreenWidth, &h.ScreenHeight, + &h.CountryCode, &h.Region, &h.City, + ); err != nil { + return nil, err + } + out = append(out, h) + } + return out, rows.Err() +} + +// RetentionCutoffs lists every workspace's prune boundary from its own window. +func (r *websiteTrackingRepository) RetentionCutoffs(ctx context.Context, now time.Time) ([]models.WebsiteTrackingRetentionCutoff, error) { + rows, err := r.db.Query(ctx, ` + SELECT organization_id, $1::timestamptz - make_interval(days => retention_days) + FROM website_tracking_settings + `, now) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []models.WebsiteTrackingRetentionCutoff + for rows.Next() { + var c models.WebsiteTrackingRetentionCutoff + if err := rows.Scan(&c.OrganizationID, &c.Before); err != nil { + return nil, err + } + out = append(out, c) + } + return out, rows.Err() +} + +// PruneBefore drops one workspace's page views older than the cutoff, then the +// browser records that have nothing left and have not been seen since. +func (r *websiteTrackingRepository) PruneBefore(ctx context.Context, orgID uuid.UUID, before time.Time) (int64, error) { + tag, err := r.db.Exec(ctx, ` + DELETE FROM website_page_hits WHERE organization_id = $1 AND occurred_at < $2 + `, orgID, before) + if err != nil { + return 0, err + } + _, err = r.db.Exec(ctx, ` + DELETE FROM website_visitors v + WHERE v.organization_id = $1 + AND v.last_seen_at < $2 + AND NOT EXISTS (SELECT 1 FROM website_page_hits h WHERE h.visitor_id = v.id) + `, orgID, before) + return tag.RowsAffected(), err +} diff --git a/realtime/lib/realtime_web/channels/org_channel.ex b/realtime/lib/realtime_web/channels/org_channel.ex index 12541e19..c717c886 100644 --- a/realtime/lib/realtime_web/channels/org_channel.ex +++ b/realtime/lib/realtime_web/channels/org_channel.ex @@ -475,8 +475,8 @@ defmodule RealtimeWeb.OrgChannel do ] -> has.(:view_campaigns) - # Contact changes - String.contains?(event_type, "CONTACT") -> + # Contact changes, and website page views tied to a contact + String.contains?(event_type, "CONTACT") or String.contains?(event_type, "PAGE_HIT") -> has.(:view_contacts) # AI contact research progress: findings are about contacts. diff --git a/tracking/src/config.rs b/tracking/src/config.rs index d2b30dfc..52b70a26 100644 --- a/tracking/src/config.rs +++ b/tracking/src/config.rs @@ -55,6 +55,9 @@ pub struct Config { pub internal_api_token: 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 + /// budget: a person does not view a page a second, a script does. + pub pagehit_rate_limit_per_min: u32, } impl Config { @@ -167,6 +170,15 @@ impl Config { .unwrap_or(300); info!("Per-source rate limit: {}/min", rate_limit_per_min); + let pagehit_rate_limit_per_min: u32 = env::var("TRACKING_PAGEHIT_RATE_LIMIT_PER_MIN") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(60); + info!( + "Per-source page-hit rate limit: {}/min", + pagehit_rate_limit_per_min + ); + Ok(Self { env: env_name, host, @@ -184,6 +196,7 @@ impl Config { backend_internal_url, internal_api_token, rate_limit_per_min, + pagehit_rate_limit_per_min, }) } @@ -258,6 +271,7 @@ impl Config { backend_internal_url, internal_api_token, rate_limit_per_min: 300, + pagehit_rate_limit_per_min: 60, }) } diff --git a/tracking/src/handlers.rs b/tracking/src/handlers.rs index 705dc57f..be2b7731 100644 --- a/tracking/src/handlers.rs +++ b/tracking/src/handlers.rs @@ -1,4 +1,5 @@ use axum::{ + body::Bytes, extract::{Path, State}, http::{header, HeaderMap, StatusCode}, response::{IntoResponse, Redirect, Response}, @@ -12,6 +13,7 @@ use std::time::Duration; use crate::abuse::{is_prefetch, is_scanner, RateLimiter}; use crate::config::Config; use crate::events::TrackingEvent; +use crate::hits::{ForwardedHit, HitForwarder, HitPayload, Outcome}; use crate::links::{LinkResolver, Resolution}; use crate::producer::Producer; @@ -36,6 +38,10 @@ pub struct AppState { pub rate_limiter: Arc, /// Click-ticket resolver (backend internal API + layered caches) pub links: Arc, + /// Website page-view forwarder (backend internal API + layered caches) + pub hits: Arc, + /// Tighter per-source budget for page views than for pixels + pub hit_rate_limiter: Arc, } impl AppState { @@ -58,6 +64,11 @@ impl AppState { config.backend_internal_url.clone(), config.internal_api_token.clone(), )), + hits: Arc::new(HitForwarder::new( + config.backend_internal_url.clone(), + config.internal_api_token.clone(), + )), + hit_rate_limiter: Arc::new(RateLimiter::new(config.pagehit_rate_limit_per_min)), } } @@ -192,14 +203,25 @@ pub async fn track_click( .map(|s| s.to_string()); // Security gateways and link previewers follow every URL in a message; - // serve them the destination but never count a click. + // serve them the destination but never count a click, and never hand + // them the identification ticket. if is_prefetch(&headers) || is_scanner(user_agent.as_deref()) { return Redirect::temporary(&link.destination).into_response(); } + // When the workspace registered the destination's host for website + // tracking, the ticket rides along so the snippet can tie the browser to + // the recipient. The ticket is opaque and per-recipient: it names no + // destination and no secret, only "the click the backend already knows". + let target = if link.identify { + with_identify_param(&link.destination, &link_id) + } else { + link.destination.clone() + }; + // Dedupe repeat clicks of the same ticket from the same source if state.is_duplicate("CLICK", &link_id, &ip_hash).await { - return Redirect::temporary(&link.destination).into_response(); + return Redirect::temporary(&target).into_response(); } // Publish event asynchronously (fire and forget) @@ -218,7 +240,154 @@ pub async fn track_click( .await; }); - Redirect::temporary(&link.destination).into_response() + Redirect::temporary(&target).into_response() +} + +/// Query parameter the click redirect appends and the snippet strips. +const IDENTIFY_PARAM: &str = "wbly_t"; + +/// Appends the identification ticket to a destination, keeping any existing +/// query and fragment intact. +fn with_identify_param(destination: &str, ticket: &str) -> String { + let (base, fragment) = match destination.split_once('#') { + Some((b, f)) => (b, Some(f)), + None => (destination, None), + }; + let sep = if base.contains('?') { '&' } else { '?' }; + let mut out = format!("{}{}{}={}", base, sep, IDENTIFY_PARAM, ticket); + if let Some(f) = fragment { + out.push('#'); + out.push_str(f); + } + out +} + +/// The tracking snippet customers embed. +/// GET /tracking.js +pub async fn tracking_js() -> Response { + ( + StatusCode::OK, + [ + ( + header::CONTENT_TYPE, + "application/javascript; charset=utf-8", + ), + (header::CACHE_CONTROL, "public, max-age=3600"), + ], + TRACKING_JS, + ) + .into_response() +} + +const TRACKING_JS: &str = include_str!("../static/tracking.js"); + +/// Website page-view ingest. +/// POST /p (JSON body, sent as text/plain so browsers skip the preflight) +/// +/// Same controls as the pixel, then the payload is validated and forwarded to +/// the backend, which owns consent policy, enrichment and storage. Nothing in +/// the URL, and nothing in the body names a contact. +pub async fn track_page_hit( + State(state): State, + headers: HeaderMap, + body: Bytes, +) -> Response { + let ip = extract_ip(&headers); + let ip_hash = ip.as_deref().map(hash_ip); + let source = ip_hash.clone().unwrap_or_else(|| "unknown".to_string()); + + // 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 + // throttles the same source's pixels and clicks. + if !state.rate_limiter.allow(&source).await || !state.hit_rate_limiter.allow(&source).await { + return (StatusCode::TOO_MANY_REQUESTS, "Slow down").into_response(); + } + + let user_agent = headers + .get(header::USER_AGENT) + .and_then(|h| h.to_str().ok()) + .map(|s| s.to_string()); + + // Prefetches, crawlers, and browsers signalling Global Privacy Control + // are acknowledged and never counted. + if is_prefetch(&headers) || is_scanner(user_agent.as_deref()) || has_gpc(&headers) { + return StatusCode::NO_CONTENT.into_response(); + } + + let payload: HitPayload = match serde_json::from_slice(&body) { + Ok(p) => p, + Err(_) => return (StatusCode::BAD_REQUEST, "Invalid payload").into_response(), + }; + if !payload.validate() { + return (StatusCode::BAD_REQUEST, "Invalid payload").into_response(); + } + + // Reloads and double-fires: acknowledged, not counted. + if state.hits.is_duplicate(&payload.v, &payload.u).await { + return StatusCode::NO_CONTENT.into_response(); + } + + let origin_host = headers + .get(header::ORIGIN) + .or_else(|| headers.get(header::REFERER)) + .and_then(|h| h.to_str().ok()) + .map(host_of) + .unwrap_or_default(); + + let hit = ForwardedHit { + site_key: payload.k, + visitor_key: payload.v, + session_key: payload.s, + consent: payload.c, + identify_token: payload.t, + url: payload.u, + title: payload.ti, + referrer: payload.r, + language: payload.l, + timezone: payload.tz, + screen_width: payload.sw, + screen_height: payload.sh, + landing: payload.ld, + user_agent: user_agent.unwrap_or_default(), + ip: ip.unwrap_or_default(), + origin_host, + }; + + match state.hits.forward(hit, &source).await { + Outcome::Accepted(None) => StatusCode::NO_CONTENT.into_response(), + Outcome::Accepted(Some(vid)) => ( + StatusCode::OK, + [(header::CONTENT_TYPE, "application/json")], + format!("{{\"vid\":\"{}\"}}", vid), + ) + .into_response(), + // An unknown key is not told apart from a declined hit: the browser + // learns nothing about which keys exist. + Outcome::UnknownSite => StatusCode::NO_CONTENT.into_response(), + Outcome::Malformed => (StatusCode::BAD_REQUEST, "Invalid payload").into_response(), + Outcome::Unavailable => { + (StatusCode::SERVICE_UNAVAILABLE, "Try again shortly").into_response() + } + } +} + +/// Sec-GPC: 1 is the browser's own opt-out signal. +fn has_gpc(headers: &HeaderMap) -> bool { + headers + .get("sec-gpc") + .and_then(|h| h.to_str().ok()) + .map(|v| v.trim() == "1") + .unwrap_or(false) +} + +/// Bare host of an Origin/Referer value ("https://www.example.com/x" -> +/// "www.example.com"). +fn host_of(value: &str) -> String { + let rest = value.split_once("://").map(|(_, r)| r).unwrap_or(value); + rest.split(['/', '?', '#']) + .next() + .unwrap_or("") + .to_ascii_lowercase() } /// Return the transparent pixel response @@ -238,8 +407,14 @@ fn pixel_response() -> Response { /// Extract and hash IP address for privacy fn extract_ip_hash(headers: &HeaderMap) -> Option { + extract_ip(headers).map(|ip| hash_ip(&ip)) +} + +/// The client IP as the proxy in front reported it. Used raw only for the +/// page-view forward, where the backend turns it into a location and drops it. +fn extract_ip(headers: &HeaderMap) -> Option { // Try various headers for the real IP - let ip = headers + headers .get("x-forwarded-for") .and_then(|h| h.to_str().ok()) .and_then(|s| s.split(',').next()) @@ -255,13 +430,36 @@ fn extract_ip_hash(headers: &HeaderMap) -> Option { .get("cf-connecting-ip") .and_then(|h| h.to_str().ok()) .map(|s| s.to_string()) - }); - - ip.map(|ip| { - // Hash the IP for privacy - let mut hasher = Sha256::new(); - hasher.update(ip.as_bytes()); - let result = hasher.finalize(); - format!("{:x}", result)[..16].to_string() // Take first 16 chars - }) + }) +} + +fn hash_ip(ip: &str) -> String { + // Hash the IP for privacy + let mut hasher = Sha256::new(); + hasher.update(ip.as_bytes()); + let result = hasher.finalize(); + format!("{:x}", result)[..16].to_string() // Take first 16 chars +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn identify_param_keeps_query_and_fragment() { + assert_eq!( + with_identify_param("https://x.com/p", "abc"), + "https://x.com/p?wbly_t=abc" + ); + assert_eq!( + with_identify_param("https://x.com/p?a=1#top", "abc"), + "https://x.com/p?a=1&wbly_t=abc#top" + ); + } + + #[test] + fn host_of_strips_scheme_and_path() { + assert_eq!(host_of("https://WWW.Example.com/a?b#c"), "www.example.com"); + assert_eq!(host_of("example.com"), "example.com"); + } } diff --git a/tracking/src/hits.rs b/tracking/src/hits.rs new file mode 100644 index 00000000..eafe8fa4 --- /dev/null +++ b/tracking/src/hits.rs @@ -0,0 +1,342 @@ +//! Website page-view ingest. +//! +//! The snippet POSTs a small JSON document; this module validates it, keeps +//! everything a browser could lie about out of the forwarded payload, and +//! hands the hit to the backend's internal API, where the user agent and IP +//! are turned into device and location and the row is stored. The same +//! layered defenses as the click resolver keep a key-spray away from the +//! backend: +//! +//! 1. negative cache: site keys confirmed unknown are dropped from memory +//! 2. per-source miss budget: real snippets never miss, so a source +//! accumulating unknown keys is probing and gets cut off +//! 3. circuit breaker: when the backend errors, forwarding pauses for a +//! cooldown instead of piling on +//! +//! Nothing in the request can name a contact. The only link between a +//! browser and a person is the click ticket the redirect appended, and the +//! backend checks that ticket against the workspace that owns the site key. + +use moka::future::Cache; +use serde::{Deserialize, Serialize}; +use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use tracing::warn; + +/// Largest request body accepted, in bytes. A page view is a few hundred. +pub const MAX_BODY_BYTES: usize = 8 * 1024; + +const MAX_URL: usize = 2048; +const MAX_TITLE: usize = 512; +const MAX_REFERRER: usize = 2048; +const MAX_LANGUAGE: usize = 32; +const MAX_TIMEZONE: usize = 64; +const MAX_KEY: usize = 64; +const MIN_KEY: usize = 16; +const MAX_SCREEN: u32 = 20_000; + +/// Consecutive backend failures before the breaker opens. +const BREAKER_TRIP: u32 = 5; +/// How long the breaker stays open once tripped. +const BREAKER_COOLDOWN: Duration = Duration::from_secs(15); +/// Unknown-site-key hits allowed per source per minute. +const MISS_BUDGET_PER_MIN: u32 = 12; +/// Repeat views of one URL by one browser inside this window are not +/// counted: double-fires, reloads, and back/forward cache restores. +const DEDUPE_WINDOW: Duration = Duration::from_secs(30); + +/// The snippet's wire format. Short keys keep the beacon small. +#[derive(Deserialize)] +pub struct HitPayload { + /// site key + pub k: String, + /// visitor id + pub v: String, + /// session id + pub s: String, + /// consent: "granted" | "implicit" + #[serde(default)] + pub c: String, + /// identification ticket from the landing URL, if any + #[serde(default)] + pub t: String, + pub u: String, + #[serde(default)] + pub ti: String, + #[serde(default)] + pub r: String, + #[serde(default)] + pub l: String, + #[serde(default)] + pub tz: String, + #[serde(default)] + pub sw: u32, + #[serde(default)] + pub sh: u32, + #[serde(default)] + pub ld: bool, +} + +fn valid_key(k: &str) -> bool { + (MIN_KEY..=MAX_KEY).contains(&k.len()) + && k.bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_') +} + +impl HitPayload { + /// Structural validation only; policy (consent, allowed hosts) is the + /// backend's call because it owns the settings. + pub fn validate(&self) -> bool { + valid_key(&self.k) + && valid_key(&self.v) + && valid_key(&self.s) + && !self.u.is_empty() + && self.u.len() <= MAX_URL + && (self.u.starts_with("http://") || self.u.starts_with("https://")) + && self.ti.len() <= MAX_TITLE + && self.r.len() <= MAX_REFERRER + && self.l.len() <= MAX_LANGUAGE + && self.tz.len() <= MAX_TIMEZONE + && self.sw <= MAX_SCREEN + && self.sh <= MAX_SCREEN + && self.t.len() <= MAX_KEY + && self.c.len() <= 16 + } +} + +/// What the backend receives: the payload plus the request facts only this +/// edge saw. Field names match the Go models.WebsiteHitRequest. +#[derive(Serialize)] +pub struct ForwardedHit { + pub site_key: String, + pub visitor_key: String, + pub session_key: String, + pub consent: String, + pub identify_token: String, + pub url: String, + pub title: String, + pub referrer: String, + pub language: String, + pub timezone: String, + pub screen_width: u32, + pub screen_height: u32, + pub landing: bool, + pub user_agent: String, + pub ip: String, + pub origin_host: String, +} + +#[derive(Deserialize, Default)] +pub struct HitResponse { + #[serde(default)] + pub new_visitor_key: String, +} + +pub enum Outcome { + /// Stored (or quietly declined by policy); optionally the browser must + /// adopt a new visitor id. + Accepted(Option), + /// Site key confirmed unknown, or this source exhausted its miss budget. + UnknownSite, + /// Payload the backend refused as malformed. + Malformed, + /// Backend unavailable / breaker open. Nothing was stored. + Unavailable, +} + +pub struct HitForwarder { + http: reqwest::Client, + backend_url: String, + internal_token: String, + unknown_sites: Cache, + miss_budget: Cache>, + dedupe: Cache, + breaker_failures: AtomicU32, + breaker_open_until_ms: AtomicU64, + started: Instant, +} + +impl HitForwarder { + pub fn new(backend_url: String, internal_token: String) -> Self { + Self { + http: reqwest::Client::builder() + .timeout(Duration::from_secs(3)) + .build() + .expect("reqwest client"), + backend_url: backend_url.trim_end_matches('/').to_string(), + internal_token, + // Short negative TTL: a key rotated or enabled seconds ago must + // start working without a restart. + unknown_sites: Cache::builder() + .max_capacity(100_000) + .time_to_live(Duration::from_secs(60)) + .build(), + miss_budget: Cache::builder() + .max_capacity(50_000) + .time_to_live(Duration::from_secs(60)) + .build(), + dedupe: Cache::builder() + .max_capacity(200_000) + .time_to_live(DEDUPE_WINDOW) + .build(), + breaker_failures: AtomicU32::new(0), + breaker_open_until_ms: AtomicU64::new(0), + started: Instant::now(), + } + } + + /// True when this browser already reported this URL inside the window. + pub async fn is_duplicate(&self, visitor: &str, url: &str) -> bool { + let key = format!("{}:{}", visitor, url); + if self.dedupe.contains_key(&key) { + return true; + } + self.dedupe.insert(key, ()).await; + false + } + + pub async fn forward(&self, hit: ForwardedHit, source: &str) -> Outcome { + if self.unknown_sites.contains_key(&hit.site_key) { + self.count_miss(source).await; + return Outcome::UnknownSite; + } + if !self.miss_allowed(source).await { + return Outcome::UnknownSite; + } + if self.breaker_is_open() { + return Outcome::Unavailable; + } + + let url = format!("{}/api/v1/internal/page-hits", self.backend_url); + let response = self + .http + .post(&url) + .bearer_auth(&self.internal_token) + .json(&hit) + .send() + .await; + + match response { + Ok(resp) if resp.status() == reqwest::StatusCode::NO_CONTENT => { + self.breaker_failures.store(0, Ordering::Relaxed); + Outcome::Accepted(None) + } + Ok(resp) if resp.status().is_success() => { + self.breaker_failures.store(0, Ordering::Relaxed); + let body = resp.json::().await.unwrap_or_default(); + let rotate = if body.new_visitor_key.is_empty() { + None + } else { + Some(body.new_visitor_key) + }; + Outcome::Accepted(rotate) + } + Ok(resp) if resp.status() == reqwest::StatusCode::NOT_FOUND => { + self.breaker_failures.store(0, Ordering::Relaxed); + self.unknown_sites.insert(hit.site_key.clone(), ()).await; + self.count_miss(source).await; + Outcome::UnknownSite + } + Ok(resp) if resp.status() == reqwest::StatusCode::BAD_REQUEST => { + self.breaker_failures.store(0, Ordering::Relaxed); + Outcome::Malformed + } + Ok(resp) => { + warn!("page-hit forward unexpected status: {}", resp.status()); + self.record_failure(); + Outcome::Unavailable + } + Err(e) => { + warn!("page-hit forward failed: {}", e); + self.record_failure(); + Outcome::Unavailable + } + } + } + + async fn miss_allowed(&self, source: &str) -> bool { + let counter = self + .miss_budget + .get_with(source.to_string(), async { Arc::new(AtomicU32::new(0)) }) + .await; + counter.load(Ordering::Relaxed) < MISS_BUDGET_PER_MIN + } + + async fn count_miss(&self, source: &str) { + let counter = self + .miss_budget + .get_with(source.to_string(), async { Arc::new(AtomicU32::new(0)) }) + .await; + counter.fetch_add(1, Ordering::Relaxed); + } + + fn now_ms(&self) -> u64 { + self.started.elapsed().as_millis() as u64 + } + + fn breaker_is_open(&self) -> bool { + self.now_ms() < self.breaker_open_until_ms.load(Ordering::Relaxed) + } + + fn record_failure(&self) { + let failures = self.breaker_failures.fetch_add(1, Ordering::Relaxed) + 1; + if failures >= BREAKER_TRIP { + self.breaker_open_until_ms.store( + self.now_ms() + BREAKER_COOLDOWN.as_millis() as u64, + Ordering::Relaxed, + ); + self.breaker_failures.store(0, Ordering::Relaxed); + warn!("page-hit breaker open for {:?}", BREAKER_COOLDOWN); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn payload(url: &str) -> HitPayload { + HitPayload { + k: "0123456789abcdef0123456789abcdef".into(), + v: "0123456789abcdef0123456789abcdef".into(), + s: "0123456789abcdef".into(), + c: "granted".into(), + t: String::new(), + u: url.into(), + ti: String::new(), + r: String::new(), + l: String::new(), + tz: String::new(), + sw: 0, + sh: 0, + ld: true, + } + } + + #[test] + fn accepts_a_plain_page_view() { + assert!(payload("https://example.com/pricing").validate()); + } + + #[test] + fn refuses_non_http_urls_and_short_keys() { + assert!(!payload("javascript:alert(1)").validate()); + let mut p = payload("https://example.com/"); + p.v = "short".into(); + assert!(!p.validate()); + let mut p = payload("https://example.com/"); + p.k = "has spaces in it!!".into(); + assert!(!p.validate()); + } + + #[test] + fn refuses_oversized_fields() { + let mut p = payload("https://example.com/"); + p.ti = "x".repeat(MAX_TITLE + 1); + assert!(!p.validate()); + let mut p = payload("https://example.com/"); + p.sw = MAX_SCREEN + 1; + assert!(!p.validate()); + } +} diff --git a/tracking/src/links.rs b/tracking/src/links.rs index 39734e79..a850a120 100644 --- a/tracking/src/links.rs +++ b/tracking/src/links.rs @@ -22,12 +22,17 @@ use tracing::warn; pub struct ResolvedLink { pub destination: String, pub task_id: String, + /// The workspace runs website tracking and registered the destination's + /// host, so the redirect may append the ticket for the snippet. + pub identify: bool, } #[derive(Deserialize)] struct LinkResponse { destination: String, task_id: String, + #[serde(default)] + identify: bool, } pub enum Resolution { @@ -127,6 +132,7 @@ impl LinkResolver { let link = ResolvedLink { destination: body.destination, task_id: body.task_id, + identify: body.identify, }; self.found.insert(link_id.to_string(), link.clone()).await; Resolution::Found(link) diff --git a/tracking/src/main.rs b/tracking/src/main.rs index d18b8a4f..b2d7be35 100644 --- a/tracking/src/main.rs +++ b/tracking/src/main.rs @@ -3,6 +3,7 @@ mod aws; mod config; mod events; mod handlers; +mod hits; #[cfg(feature = "kafka")] mod kafka; mod links; @@ -10,7 +11,11 @@ mod nats; mod observability; mod producer; -use axum::{routing::get, Router}; +use axum::{ + extract::DefaultBodyLimit, + routing::{get, post}, + Router, +}; use std::net::SocketAddr; use std::time::Duration; use tower_http::{ @@ -21,7 +26,7 @@ use tracing::{info, warn}; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; use crate::config::Config; -use crate::handlers::{health, track_click, track_open, AppState}; +use crate::handlers::{health, track_click, track_open, track_page_hit, tracking_js, AppState}; use crate::observability::report_error; use crate::producer::Producer; @@ -96,6 +101,13 @@ async fn main() { .route("/health", get(health)) .route("/t/o/:task_id", get(track_open)) .route("/c/:link_id", get(track_click)) + // Website tracking: the snippet and the page-view ingest it posts to. + // The body cap is the first thing a flood hits. + .route("/tracking.js", get(tracking_js)) + .route( + "/p", + post(track_page_hit).layer(DefaultBodyLimit::max(hits::MAX_BODY_BYTES)), + ) .layer( CorsLayer::new() .allow_origin(Any) diff --git a/tracking/static/tracking.js b/tracking/static/tracking.js new file mode 100644 index 00000000..72b49db0 --- /dev/null +++ b/tracking/static/tracking.js @@ -0,0 +1,127 @@ +/* Warmbly website tracking. Records page views for the workspace whose site + * key the script tag carries, sends nothing before consent when the + * workspace requires it, and honours Global Privacy Control / Do Not Track. + * API: warmbly('consent','granted'|'denied'), warmbly('page'), warmbly('reset'). */ +(function () { + if (window.__wbly) return; + window.__wbly = 1; + var w = window, d = document, n = navigator; + var sc = d.currentScript; + if (!sc) { + var all = d.getElementsByTagName('script'); + for (var i = all.length - 1; i >= 0; i--) { + if (all[i].getAttribute('data-site')) { sc = all[i]; break; } + } + } + if (!sc) return; + var key = sc.getAttribute('data-site'); + if (!key) return; + var mode = sc.getAttribute('data-consent') === 'implicit' ? 'implicit' : 'explicit'; + var base = sc.getAttribute('data-endpoint') || String(sc.src).replace(/\/tracking\.js.*$/, ''); + var endpoint = base + '/p'; + var VID = 'wbly_vid', SID = 'wbly_sid', SAT = 'wbly_sat', CON = 'wbly_consent', TOK = 'wbly_t'; + + function rnd() { + var a = new Uint8Array(16); + if (w.crypto && w.crypto.getRandomValues) { w.crypto.getRandomValues(a); } + else { for (var i = 0; i < 16; i++) a[i] = (Math.random() * 256) | 0; } + var s = ''; + for (var j = 0; j < 16; j++) s += ('0' + a[j].toString(16)).slice(-2); + return s; + } + function get(k) { try { return w.localStorage.getItem(k); } catch (e) { return null; } } + function set(k, v) { try { w.localStorage.setItem(k, v); } catch (e) {} } + function del(k) { try { w.localStorage.removeItem(k); } catch (e) {} } + function sget(k) { try { return w.sessionStorage.getItem(k); } catch (e) { return null; } } + function sset(k, v) { try { w.sessionStorage.setItem(k, v); } catch (e) {} } + function sdel(k) { try { w.sessionStorage.removeItem(k); } catch (e) {} } + + function optedOut() { + return n.globalPrivacyControl === true || n.doNotTrack === '1' || w.doNotTrack === '1'; + } + function consent() { + if (optedOut()) return null; + var c = get(CON); + if (c === 'granted') return 'granted'; + if (c === 'denied') return null; + return mode === 'implicit' ? 'implicit' : null; + } + + /* The click redirect appends the identification ticket; take it off the + * address bar right away so it is never bookmarked or shared. */ + var token = ''; + try { + var u = new URL(w.location.href); + var t = u.searchParams.get(TOK); + if (t) { + token = t; + u.searchParams.delete(TOK); + w.history.replaceState(w.history.state, '', u.toString()); + } + } catch (e) {} + + var landing = false, lastUrl = '', lastAt = 0; + function session() { + var now = Date.now(); + var sid = sget(SID), at = +sget(SAT) || 0; + if (!sid || now - at > 30 * 60 * 1000) { sid = rnd(); landing = true; } + sset(SID, sid); + sset(SAT, String(now)); + return sid; + } + function tz() { + try { return Intl.DateTimeFormat().resolvedOptions().timeZone || ''; } catch (e) { return ''; } + } + + function send() { + var c = consent(); + if (!c) return; + var url = w.location.href, now = Date.now(); + if (url === lastUrl && now - lastAt < 1000) return; + lastUrl = url; lastAt = now; + var vid = get(VID); + if (!vid) { vid = rnd(); set(VID, vid); } + var sid = session(); + var body = { + k: key, v: vid, s: sid, c: c, t: token, u: url, + ti: d.title || '', r: d.referrer || '', l: n.language || '', tz: tz(), + sw: (w.screen && w.screen.width) || 0, sh: (w.screen && w.screen.height) || 0, ld: landing + }; + landing = false; token = ''; + try { + /* text/plain keeps this a simple request: no preflight on every view. */ + fetch(endpoint, { + method: 'POST', keepalive: true, credentials: 'omit', + headers: { 'Content-Type': 'text/plain' }, body: JSON.stringify(body) + }).then(function (r) { return r.status === 200 ? r.json() : null; }) + .then(function (j) { if (j && j.vid) set(VID, j.vid); }) + .catch(function () {}); + } catch (e) {} + } + + /* Single-page apps: a pushed history entry is a page view. */ + var push = w.history.pushState; + if (push) { + w.history.pushState = function () { + push.apply(this, arguments); + setTimeout(send, 0); + }; + } + w.addEventListener('popstate', function () { setTimeout(send, 0); }); + + var queued = (w.warmbly && w.warmbly.q) || []; + function api(cmd, arg) { + if (cmd === 'consent') { + if (arg === 'granted') { set(CON, 'granted'); send(); } + else if (arg === 'denied') { set(CON, 'denied'); del(VID); sdel(SID); sdel(SAT); } + } else if (cmd === 'reset') { + del(VID); sdel(SID); sdel(SAT); + } else if (cmd === 'page') { + send(); + } + } + w.warmbly = function () { api.apply(null, arguments); }; + w.warmbly.q = []; + for (var q = 0; q < queued.length; q++) api.apply(null, queued[q]); + send(); +})(); diff --git a/web/src/app/app/settings/layout.tsx b/web/src/app/app/settings/layout.tsx index 45c0414b..407b602a 100644 --- a/web/src/app/app/settings/layout.tsx +++ b/web/src/app/app/settings/layout.tsx @@ -29,6 +29,7 @@ import { UserIcon, UsersIcon, WebhookIcon, + GlobeIcon, } from "lucide-react"; import { UnsavedProvider, useUnsavedRegistry } from "@/hooks/context/unsaved"; import { usePermission, type PermissionKey } from "@/hooks/usePermission"; @@ -68,6 +69,7 @@ const GROUPS: SectionGroup[] = [ { path: "roles", label: "Roles & access", icon: ShieldCheckIcon, description: "Who can do what.", ownerOnly: true }, { path: "workspace", label: "Workspace", icon: BriefcaseIcon, description: "Org-wide settings.", ownerOnly: true }, { path: "sending", label: "Sending", icon: SendIcon, description: "When campaign mail reaches each recipient.", permission: "MANAGE_SETTINGS" }, + { path: "tracking", label: "Website tracking", icon: GlobeIcon, description: "Page views on your site, in the contact timeline.", permission: "MANAGE_SETTINGS" }, { path: "ai-skills", label: "AI skills", icon: SparklesIcon, description: "Playbooks your AI features follow.", permission: "MANAGE_SETTINGS" }, { path: "billing", label: "Billing", icon: CreditCardIcon, description: "Plan, payment, invoices.", ownerOnly: true, billingOnly: true }, { path: "referral", label: "Refer & earn", icon: GiftIcon, description: "Invite teams and earn account credit.", ownerOnly: true, billingOnly: true }, diff --git a/web/src/app/app/settings/tracking/page.tsx b/web/src/app/app/settings/tracking/page.tsx new file mode 100644 index 00000000..1c9229b9 --- /dev/null +++ b/web/src/app/app/settings/tracking/page.tsx @@ -0,0 +1,289 @@ +// Website tracking settings: the snippet a workspace installs on its own +// site, and the privacy posture it runs under. Off by default, and the +// consent mode, location precision and retention window are all the +// workspace's decision, enforced on the server rather than in the snippet. + +import React from "react"; +import { CheckIcon, CopyIcon } from "lucide-react"; +import { Row, Section, SectionShell, Toggle } from "../_components/SectionShell"; +import { NoAccess } from "@/components/layout/NoAccess"; +import { usePermission } from "@/hooks/usePermission"; +import { useConfirm } from "@/hooks/context/confirm"; +import SaveStatus from "../_components/SaveStatus"; +import { SelectMenu, type SelectOption } from "@/components/ui/select-menu"; +import { NumberInput } from "@/components/ui/field"; +import { Textarea } from "@/components/ui/textarea"; +import { useAutosave } from "@/hooks/useAutosave"; +import { useRegisterUnsaved } from "@/hooks/context/unsaved"; +import { + useRotateWebsiteTrackingKey, + useUpdateWebsiteTrackingSettings, + useWebsiteTrackingSettings, +} from "@/lib/api/hooks/app/websitetracking/useWebsiteTracking"; +import { + WEBSITE_RETENTION_MAX_DAYS, + WEBSITE_RETENTION_MIN_DAYS, + trackingSnippet, + type UpdateWebsiteTrackingSettings, + type WebsiteTrackingSettings, +} from "@/lib/api/models/app/websitetracking/WebsiteTrackingSettings"; + +const CONSENT_OPTIONS: SelectOption[] = [ + { value: "explicit", label: "Ask first (recommended)" }, + { value: "implicit", label: "Record on load" }, +]; + +const LOCATION_OPTIONS: SelectOption[] = [ + { value: "none", label: "Do not keep" }, + { value: "country", label: "Country only" }, + { value: "city", label: "Country, region and city" }, +]; + +type Draft = Pick< + WebsiteTrackingSettings, + "enabled" | "consent_mode" | "location_precision" | "retention_days" +> & { hosts: string }; + +function toDraft(s: WebsiteTrackingSettings): Draft { + return { + enabled: s.enabled, + consent_mode: s.consent_mode, + location_precision: s.location_precision, + retention_days: s.retention_days, + hosts: s.allowed_hosts.join("\n"), + }; +} + +function toPatch(d: Draft): UpdateWebsiteTrackingSettings { + return { + enabled: d.enabled, + consent_mode: d.consent_mode, + location_precision: d.location_precision, + retention_days: d.retention_days, + allowed_hosts: d.hosts + .split(/[\n,]/) + .map((h) => h.trim()) + .filter(Boolean), + }; +} + +export default function WebsiteTrackingSettingsPage() { + const canManage = usePermission("MANAGE_SETTINGS"); + if (!canManage) return ; + return ; +} + +function WebsiteTrackingSettingsView() { + const { data, isLoading } = useWebsiteTrackingSettings(); + const update = useUpdateWebsiteTrackingSettings(); + const rotate = useRotateWebsiteTrackingKey(); + const confirm = useConfirm(); + const [draft, setDraft] = React.useState(null); + + const autosave = useAutosave({ + value: draft, + enabled: !!draft, + debounceMs: 600, + save: async (v) => { + if (v) await update.mutateAsync(toPatch(v)); + }, + }); + useRegisterUnsaved(autosave, () => setDraft(autosave.savedValue)); + + // One-shot hydration, as on the other autosave settings pages: the server + // seeds the draft once and the save path owns the baseline after that. + const hydrated = React.useRef(false); + React.useEffect(() => { + if (!data || hydrated.current) return; + hydrated.current = true; + const d = toDraft(data); + setDraft(d); + autosave.markSaved(d); + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [data]); + + const patch = React.useCallback((next: Partial) => { + setDraft((prev) => (prev ? { ...prev, ...next } : prev)); + }, []); + + const hostCount = draft ? toPatch(draft).allowed_hosts?.length ?? 0 : 0; + + return ( + } + > +
+ {isLoading || !draft ? ( +
+ ) : ( + <> + + patch({ enabled: on })} /> + + + + patch({ consent_mode: v as Draft["consent_mode"] })} + options={CONSENT_OPTIONS} + aria-label="Consent mode" + minWidth={220} + align="end" + /> + + + + patch({ location_precision: v as Draft["location_precision"] })} + options={LOCATION_OPTIONS} + aria-label="Location precision" + minWidth={220} + align="end" + /> + + + + + patch({ + retention_days: Number.isFinite(n) + ? Math.min(WEBSITE_RETENTION_MAX_DAYS, Math.max(WEBSITE_RETENTION_MIN_DAYS, n)) + : 90, + }) + } + className="w-24" + /> + + + +
+