Architecture spec for the per-lead daily verification cap, with HubSpot as the sales/support edit surface and the TokenID backend as the runtime source of truth.
Sales and support need to bump an individual lead's daily verification cap without filing engineering tickets. The verifier hot path must not depend on HubSpot availability.
┌─────────────────────┐ contact.propertyChange ┌─────────────────────┐
│ HubSpot Contact │ ───── webhook (HMAC) ──────▶ │ TokenID backend │
│ custom property: │ │ /webhooks/hubspot │
│ cap_events_per_day │ │ │
└─────────────────────┘ │ writes: │
│ Organization │
│ .cap_events_per_day
│ + cap_changes │
│ audit row │
└──────────┬──────────┘
│
▼
┌─────────────────────┐
│ /verify-free │
│ reads cap from DB │
│ (no HubSpot call) │
└─────────────────────┘
Custom contact property on the existing contact object. Sales/support edits the property on the contact record like any other field.
| Field | Value |
|---|---|
| Name | cap_events_per_day |
| Group | tokenid_entitlements (new group) |
| Type | Number, integer |
| Field type | Number |
| Description | "Daily TokenID verification limit for this lead. Empty = use tier default." |
| Read-only via API | No |
HubSpot → backend. One subscription, one event type.
Subscription: contact.propertyChange
Property filter: cap_events_per_day
Endpoint: POST https://api.tokenid.<host>/webhooks/hubspot
Signature: X-HubSpot-Signature-v3 (HMAC-SHA256 of timestamp+method+uri+body)
Webhook payload contains objectId (HubSpot contact id), propertyName,
propertyValue, changeSource, sourceId (HubSpot user id of the editor).
# app/routers/webhooks_hubspot.py
@router.post("/webhooks/hubspot", status_code=200)
async def handle_hubspot_event(
request: Request,
db: AsyncSession = Depends(get_db),
):
# 1. Verify HMAC signature
raw_body = await request.body()
if not verify_hubspot_signature(request.headers, raw_body):
raise HTTPException(401, "bad_signature")
payload = json.loads(raw_body)
for event in payload:
if event.get("propertyName") != "cap_events_per_day":
continue
contact_id = event["objectId"]
new_value = event.get("propertyValue") # str or None
# 2. Map HubSpot contact -> Organization
org = await db.scalar(
select(Organization).where(Organization.hubspot_contact_id == contact_id)
)
if not org:
logger.warning("hubspot_webhook: unknown contact %s", contact_id)
continue
# 3. Parse and validate
new_cap = int(new_value) if new_value else None
if new_cap is not None and (new_cap < 0 or new_cap > 100_000):
logger.warning("hubspot_webhook: cap %s out of range for org %s", new_cap, org.id)
continue
# 4. Write to DB + audit row in one transaction
old_cap = org.cap_events_per_day
org.cap_events_per_day = new_cap
db.add(CapChange(
organization_id=org.id,
old_value=old_cap,
new_value=new_cap,
source="hubspot_webhook",
actor_hubspot_user_id=event.get("sourceId"),
event_timestamp=datetime.utcnow(),
))
await db.commit()
logger.info("cap updated org=%s %s -> %s", org.id, old_cap, new_cap)
return {"ok": True}# Organization model (add column)
class Organization(Base):
# ... existing columns ...
cap_events_per_day: Mapped[int | None] = mapped_column(nullable=True)
hubspot_contact_id: Mapped[str | None] = mapped_column(
nullable=True, index=True, unique=True
)
# New audit table
class CapChange(Base):
__tablename__ = "cap_changes"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
organization_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("organizations.id", ondelete="CASCADE"), index=True
)
old_value: Mapped[int | None]
new_value: Mapped[int | None]
source: Mapped[str] # "hubspot_webhook" | "admin_ui" | "support_tool"
actor_hubspot_user_id: Mapped[str | None]
event_timestamp: Mapped[datetime]
created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)Alembic migration adds the column + table + the unique index on
hubspot_contact_id.
The /verify-free endpoint reads the cap from the DB, never from HubSpot:
async def get_effective_cap(org: Organization) -> int:
if org.cap_events_per_day is not None:
return org.cap_events_per_day
return TIER_DEFAULT_CAPS[org.license_tier]Where TIER_DEFAULT_CAPS is a constant dict — values are the open product
decision waiting on Deepak.
- Default per-lead daily cap for the free tier — the integer to seed in
TIER_DEFAULT_CAPS[LicenseTier.FREE]. Pro tier number too, if Pro gets a different baseline. - Authorization on the bump action. Options, in increasing order of
strictness:
- Anyone with HubSpot Contact edit access can bump
- Restrict to a named HubSpot user role (e.g. "Account Manager")
- Restrict the
cap_events_per_dayproperty write permission via HubSpot property-level permissions
If we go with option 2 or 3, the property's API permission needs to mirror the in-app restriction so the webhook signature doesn't lie.
- Alembic migration: add
cap_events_per_daycolumn toorganizations, addhubspot_contact_idcolumn + unique index, createcap_changes. - HubSpot: create the property + the
tokenid_entitlementsgroup. - Backend: implement webhook handler + signature verifier + tests.
- HubSpot: register the webhook subscription on the property.
- Backfill: for orgs that already have a HubSpot contact, populate
hubspot_contact_idfrom the existing sync. - Smoke test: bump a test contact's cap in HubSpot → confirm webhook hits
backend → confirm
cap_events_per_dayupdates → confirm /verify-free uses the new cap.
Single PR. Migration + handler + tests. ~400 LOC including the test file.
- Tier itself (
license_tier) is still set on the org at signup / upgrade. This spec only governs the per-lead numeric override on top of the tier default. requires_tierdependency continues to gate endpoints by tier. The cap governs request-count-per-day inside the tier, not feature access.- HubSpot is only consulted during the property-change webhook. The hot path stays HubSpot-free.