diff --git a/packages/sync/src/providers/google/google-event.normalizer.test.ts b/packages/sync/src/providers/google/google-event.normalizer.test.ts index 4ea643d28..1f2ce6f55 100644 --- a/packages/sync/src/providers/google/google-event.normalizer.test.ts +++ b/packages/sync/src/providers/google/google-event.normalizer.test.ts @@ -148,8 +148,8 @@ describe("normalizeGoogleEvent", () => { throw new Error("expected instance"); expect(read.recurrence.seriesProviderId).toBe("15chil19v5nskedvmo93ei4nl8"); - expect(Date.parse(read.recurrence.recurrenceId)).toBe( - Date.parse("2025-09-07T02:30:00+01:00"), + expect(read.recurrence.recurrenceId).toBe( + new Date("2025-09-07T02:30:00+01:00").toISOString(), ); }); @@ -160,9 +160,10 @@ describe("normalizeGoogleEvent", () => { "tlf9q8uk5vjl2i2868q36dpi28_20130508T220000Z", ); expect(read.series?.seriesProviderId).toBe("tlf9q8uk5vjl2i2868q36dpi28"); - // originalStartTime has no timeZone: the id must be a deterministic, - // host-independent UTC string, not one derived from the host's zone. - expect(read.series?.recurrenceId).toBe("2013-05-08T22:00:00+00:00"); + // originalStartTime has no timeZone: the id must be Compass's canonical + // UTC recurrenceId (Date#toISOString), not an offset string — otherwise a + // later scope-"this" command keyed on the projected form misses the row. + expect(read.series?.recurrenceId).toBe("2013-05-08T22:00:00.000Z"); }); it("maps a cancelled standalone event to a cancellation with no series", () => { diff --git a/packages/sync/src/providers/google/google-event.normalizer.ts b/packages/sync/src/providers/google/google-event.normalizer.ts index b005d8e0c..02563a053 100644 --- a/packages/sync/src/providers/google/google-event.normalizer.ts +++ b/packages/sync/src/providers/google/google-event.normalizer.ts @@ -82,7 +82,7 @@ function cancellationSeries( if (!item.recurringEventId || !item.originalStartTime) return null; return { seriesProviderId: item.recurringEventId, - recurrenceId: toOffsetIso(item.originalStartTime), + recurrenceId: toCanonicalRecurrenceId(item.originalStartTime), }; } @@ -249,12 +249,26 @@ function mapRecurrence(item: gSchema$Event): ProviderEventRecurrence { return { kind: "instance", seriesProviderId: item.recurringEventId, - recurrenceId: toOffsetIso(item.originalStartTime), + // Canonical UTC form — must match Compass projection / command + // recurrenceIds (Date#toISOString). An offset string for the same + // instant would miss series_exception_identity and collide + // provider_event_identity when a scope-"this" write upserts. + recurrenceId: toCanonicalRecurrenceId(item.originalStartTime), }; } return { kind: "single" }; } +// Recurrence identity as Compass mints it everywhere else: UTC ISO with +// milliseconds (Date#toISOString). Schedule start/end keep offset form via +// toOffsetIso so the wall-clock zone stays visible; only the identity key +// must be byte-identical across import, projection, and command paths. +function toCanonicalRecurrenceId( + eventDateTime: calendar_v3.Schema$EventDateTime, +): string { + return new Date(toOffsetIso(eventDateTime)).toISOString(); +} + // A Google event date-time as a deterministic RFC3339 offset string, never // dependent on the host's zone. With a zone, re-anchor to it so the offset is // correct for that date's DST rules. Without one, canonicalize to UTC rather diff --git a/packages/sync/src/storage/repositories/event.repository.db.test.ts b/packages/sync/src/storage/repositories/event.repository.db.test.ts index 8c3f29667..bc9ee408a 100644 --- a/packages/sync/src/storage/repositories/event.repository.db.test.ts +++ b/packages/sync/src/storage/repositories/event.repository.db.test.ts @@ -437,4 +437,262 @@ describe("EventRepository", () => { ); expect(exceptions).toHaveLength(1); }); + + // Regression for PostHog E11000 on provider_event_identity: import stores a + // Google instance via upsertByProviderIdentity with an offset-form + // recurrenceId, then a scope-"this" command upserts with the canonical UTC + // form of the same instant. Without converging on the provider-identity + // document, the series-keyed insert collides the unique index. + it("converges upsertException onto an imported provider-identity exception", async () => { + const tenantId = objectId(); + const principalId = objectId(); + const calendarId = objectId(); + const connectionId = objectId(); + const offsetRecurrenceId = "2026-08-10T13:00:00-06:00" as never; + const utcRecurrenceId = "2026-08-10T19:00:00.000Z" as never; + const providerEventId = "g-series_20260810T190000Z"; + + const master = await repo.put( + compassRecord({ + tenantId, + principalId, + calendarId, + connectionId, + providerEventId: "g-series", + providerVersion: "etag-master", + origin: "provider", + recurrence: { + kind: "seriesMaster", + rules: ["RRULE:FREQ=WEEKLY"], + } as EventRecord["recurrence"], + }), + ); + + const imported = await repo.upsertByProviderIdentity( + linkedUpsert({ + tenantId, + principalId, + calendarId, + connectionId, + providerEventId, + providerVersion: "etag-inst", + content: { ...baseContent, title: "Imported override" }, + schedule: timed( + "2026-08-10T14:00:00-06:00", + "2026-08-10T15:00:00-06:00", + ), + recurrence: { + kind: "exception", + seriesId: master._id, + recurrenceId: offsetRecurrenceId, + cancelled: false, + } as EventRecord["recurrence"], + }), + ); + + const updated = await repo.upsertException( + master, + utcRecurrenceId, + { + content: { ...baseContent, title: "Command override" }, + schedule: timed( + "2026-08-10T14:00:00-06:00", + "2026-08-10T15:00:00-06:00", + ), + cancelled: false, + providerIdentity: { + providerEventId: providerEventId as never, + providerVersion: "etag-inst-2" as never, + }, + }, + new Date(), + ); + + expect(updated._id).toBe(imported._id); + expect(updated.content.title).toBe("Command override"); + expect(updated.providerEventId).toBe(providerEventId); + expect(updated.providerVersion).toBe("etag-inst-2"); + if (updated.recurrence.kind === "exception") { + expect(updated.recurrence.seriesId).toBe(master._id); + expect(updated.recurrence.recurrenceId).toBe(utcRecurrenceId); + } + const exceptions = await repo.findSeriesExceptions( + tenantId, + principalId, + master._id, + ); + expect(exceptions).toHaveLength(1); + }); + + it("drops a series-keyed duplicate when converging on provider identity", async () => { + const tenantId = objectId(); + const principalId = objectId(); + const calendarId = objectId(); + const connectionId = objectId(); + const offsetRecurrenceId = "2026-08-15T14:00:00-06:00" as never; + const utcRecurrenceId = "2026-08-15T20:00:00.000Z" as never; + const providerEventId = "g-series_20260815T200000Z"; + + const master = await repo.put( + compassRecord({ + tenantId, + principalId, + calendarId, + connectionId, + providerEventId: "g-series", + providerVersion: "etag-master", + origin: "provider", + recurrence: { + kind: "seriesMaster", + rules: ["RRULE:FREQ=WEEKLY"], + } as EventRecord["recurrence"], + }), + ); + + // Series-keyed tombstone from a prior command that used the UTC form + // (instance already gone → null provider identity). + const tombstone = await repo.upsertException( + master, + utcRecurrenceId, + { + content: baseContent, + schedule: timed( + "2026-08-15T14:00:00-06:00", + "2026-08-15T15:00:00-06:00", + ), + cancelled: true, + providerIdentity: null, + }, + new Date(), + ); + + // Import still carries the live provider instance under the offset form. + const imported = await repo.upsertByProviderIdentity( + linkedUpsert({ + tenantId, + principalId, + calendarId, + connectionId, + providerEventId, + providerVersion: "etag-inst", + content: { ...baseContent, title: "Still at provider" }, + schedule: timed( + "2026-08-15T14:00:00-06:00", + "2026-08-15T15:00:00-06:00", + ), + recurrence: { + kind: "exception", + seriesId: master._id, + recurrenceId: offsetRecurrenceId, + cancelled: false, + } as EventRecord["recurrence"], + }), + ); + expect(imported._id).not.toBe(tombstone._id); + + const converged = await repo.upsertException( + master, + utcRecurrenceId, + { + content: { ...baseContent, title: "Retried delete" }, + schedule: timed( + "2026-08-15T14:00:00-06:00", + "2026-08-15T15:00:00-06:00", + ), + cancelled: true, + providerIdentity: { + providerEventId: providerEventId as never, + providerVersion: "etag-inst" as never, + }, + }, + new Date(), + ); + + expect(converged._id).toBe(imported._id); + expect(converged.content.title).toBe("Retried delete"); + if (converged.recurrence.kind === "exception") { + expect(converged.recurrence.cancelled).toBe(true); + expect(converged.recurrence.recurrenceId).toBe(utcRecurrenceId); + } + expect( + await repo.findById(tenantId, principalId, tombstone._id), + ).toBeNull(); + expect( + await repo.findSeriesExceptions(tenantId, principalId, master._id), + ).toHaveLength(1); + }); + + // After recurrenceId canonicalization, import and command share the UTC + // series_exception_identity key. A null-provider tombstone left by a + // scope-"this" delete must be adopted by the provider-identity upsert, + // not collide it. + it("adopts a series-keyed tombstone when importing a provider exception", async () => { + const tenantId = objectId(); + const principalId = objectId(); + const calendarId = objectId(); + const connectionId = objectId(); + const utcRecurrenceId = "2026-08-10T19:00:00.000Z" as never; + const providerEventId = "g-series_20260810T190000Z"; + + const master = await repo.put( + compassRecord({ + tenantId, + principalId, + calendarId, + connectionId, + providerEventId: "g-series", + providerVersion: "etag-master", + origin: "provider", + recurrence: { + kind: "seriesMaster", + rules: ["RRULE:FREQ=WEEKLY"], + } as EventRecord["recurrence"], + }), + ); + + const tombstone = await repo.upsertException( + master, + utcRecurrenceId, + { + content: baseContent, + schedule: timed( + "2026-08-10T13:00:00-06:00", + "2026-08-10T14:00:00-06:00", + ), + cancelled: true, + providerIdentity: null, + }, + new Date(), + ); + expect(tombstone.providerEventId).toBeNull(); + + const imported = await repo.upsertByProviderIdentity( + linkedUpsert({ + tenantId, + principalId, + calendarId, + connectionId, + providerEventId, + providerVersion: "etag-inst", + content: { ...baseContent, title: "Cancelled at provider" }, + schedule: timed( + "2026-08-10T13:00:00-06:00", + "2026-08-10T14:00:00-06:00", + ), + recurrence: { + kind: "exception", + seriesId: master._id, + recurrenceId: utcRecurrenceId, + cancelled: true, + } as EventRecord["recurrence"], + }), + ); + + expect(imported._id).toBe(tombstone._id); + expect(imported.providerEventId).toBe(providerEventId); + expect(imported.content.title).toBe("Cancelled at provider"); + expect( + await repo.findSeriesExceptions(tenantId, principalId, master._id), + ).toHaveLength(1); + }); }); diff --git a/packages/sync/src/storage/repositories/event.repository.ts b/packages/sync/src/storage/repositories/event.repository.ts index 2e66f5324..9ee6eb753 100644 --- a/packages/sync/src/storage/repositories/event.repository.ts +++ b/packages/sync/src/storage/repositories/event.repository.ts @@ -52,6 +52,36 @@ export class EventRepository { options?: UpsertByProviderIdentityOptions, ): Promise { const now = new Date(); + // A prior scope-"this" command may have left a series-keyed exception + // (often a null-provider tombstone) at this canonical recurrenceId. + // Adopt or drop it before the provider-identity upsert, or the insert + // collides series_exception_identity — the dual of the E11000 that + // upsertException converges the other direction. + if (input.recurrence.kind === "exception") { + await this.#reconcileSeriesExceptionBeforeProviderUpsert(input); + } + + try { + return await this.#upsertByProviderIdentityOnce(input, options, now); + } catch (error) { + if ( + !isDuplicateKeyError(error) || + input.recurrence.kind !== "exception" + ) { + throw error; + } + // Concurrent command upsert won the series key between reconcile and + // insert. Reconcile again and retry once. + await this.#reconcileSeriesExceptionBeforeProviderUpsert(input); + return this.#upsertByProviderIdentityOnce(input, options, now); + } + } + + async #upsertByProviderIdentityOnce( + input: ProviderEventUpsert, + options: UpsertByProviderIdentityOptions | undefined, + now: Date, + ): Promise { const filter = { connectionId: input.connectionId, calendarId: input.calendarId, @@ -171,6 +201,59 @@ export class EventRepository { return EventRecordSchema.parse(result); } + // Stamp provider identity onto a series-keyed exception that lacks it (or + // drop a divergent series-keyed duplicate) so the provider-identity upsert + // that follows updates one document instead of colliding + // series_exception_identity. + async #reconcileSeriesExceptionBeforeProviderUpsert( + input: ProviderEventUpsert, + ): Promise { + if (input.recurrence.kind !== "exception") return; + + const bySeries = await this.collection.findOne({ + tenantId: input.tenantId, + principalId: input.principalId, + "recurrence.kind": "exception", + "recurrence.seriesId": input.recurrence.seriesId, + "recurrence.recurrenceId": input.recurrence.recurrenceId, + }); + if (!bySeries) return; + + const byProvider = await this.collection.findOne({ + connectionId: input.connectionId, + calendarId: input.calendarId, + providerEventId: { + $eq: input.providerEventId, + $type: "string" as const, + }, + }); + + if (!byProvider) { + await this.collection.updateOne( + { + _id: bySeries._id, + tenantId: input.tenantId, + principalId: input.principalId, + }, + { + $set: { + connectionId: input.connectionId, + providerEventId: input.providerEventId, + }, + }, + ); + return; + } + + if (bySeries._id !== byProvider._id) { + await this.collection.deleteOne({ + _id: bySeries._id, + tenantId: input.tenantId, + principalId: input.principalId, + }); + } + } + // Full write of a Compass (or already-identified) event by its _id. Used for // unlinked cloud events and for promoting/relinking an existing event. The // filter is scoped to the owning tenant/principal, not _id alone: _id is the @@ -289,6 +372,14 @@ export class EventRepository { // Generation is a watermark here too (see upsertByProviderIdentity): a repair // re-seeing an exception bumps it into the new generation in place, so the // filter excludes generation and the index stays generation-free by design. + // + // Provider-linked exceptions have a second identity: provider_event_identity. + // Import writes them via upsertByProviderIdentity; commands write via this + // method. When a prior import already stored the instance under its provider + // id (possibly with a differently-formatted recurrenceId string for the same + // instant), we must converge on that document — inserting a second row with + // the same providerEventId throws E11000 on provider_event_identity and left + // staleCommandRetry looping on the failed command. async upsertException( master: EventRecord, recurrenceId: DateTime, @@ -321,48 +412,141 @@ export class EventRepository { const providerVersion = hasExplicitProviderIdentity ? (override.providerIdentity?.providerVersion ?? null) : master.providerVersion; + + const fields = { + origin: master.origin, + calendarId: master.calendarId, + clientEventId: null, + connectionId: master.connectionId, + providerEventId, + providerVersion, + providerUpdatedAt: master.providerUpdatedAt, + deliveryState: master.connectionId ? "confirmed" : master.deliveryState, + providerMetadata: master.providerMetadata, + content: override.content, + schedule: override.schedule, + "recurrence.cancelled": override.cancelled, + lifecycleState: "active" as const, + generation: master.generation, + updatedAt: now, + }; + + // Prefer the provider-identity document when one already exists (import + // path). Drop a series-keyed duplicate that lost the string-form match so + // updating recurrenceId onto the provider row cannot hit + // series_exception_identity. + if (providerEventId !== null && master.connectionId) { + const converged = await this.#convergeExceptionOntoProviderIdentity( + master, + recurrenceId, + providerEventId, + fields, + ); + if (converged) return converged; + } + + try { + const result = await this.collection.findOneAndUpdate( + { + tenantId: master.tenantId, + principalId: master.principalId, + "recurrence.kind": "exception", + "recurrence.seriesId": master._id, + "recurrence.recurrenceId": recurrenceId, + }, + { + // Mirror the master's ownership/calendar identity; set the + // instance's own content, schedule, provider identity, and cancelled + // flag. recurrence.kind/seriesId/recurrenceId are seeded from the + // filter on insert, so only cancelled is set here (setting the whole + // recurrence would conflict). + $set: fields, + $setOnInsert: { + _id: new ObjectId().toHexString() as EventId, + createdAt: now, + confirmedAt: now, + }, + }, + { upsert: true, returnDocument: "after" }, + ); + if (!result) throw new Error("Exception upsert did not return a record"); + return EventRecordSchema.parse(result); + } catch (error) { + // Concurrent import won the provider_event_identity insert between our + // lookup and this upsert. Converge on that row instead of failing the + // command (which staleCommandRetry would then loop on forever). + if ( + isDuplicateKeyError(error) && + providerEventId !== null && + master.connectionId + ) { + const converged = await this.#convergeExceptionOntoProviderIdentity( + master, + recurrenceId, + providerEventId, + fields, + ); + if (converged) return converged; + } + throw error; + } + } + + // Update the existing provider-identity row into the series exception shape + // the command wants. Removes a series-keyed duplicate first when the two + // identities diverged (offset vs UTC recurrenceId strings for one instant). + async #convergeExceptionOntoProviderIdentity( + master: EventRecord, + recurrenceId: DateTime, + providerEventId: NonNullable, + fields: Record, + ): Promise { + if (!master.connectionId) return null; + const byProvider = await this.findByProviderIdentity( + master.tenantId, + master.principalId, + { + connectionId: master.connectionId, + calendarId: master.calendarId, + providerEventId, + }, + ); + if (!byProvider) return null; + + const bySeries = await this.collection.findOne({ + tenantId: master.tenantId, + principalId: master.principalId, + "recurrence.kind": "exception", + "recurrence.seriesId": master._id, + "recurrence.recurrenceId": recurrenceId, + }); + if (bySeries && bySeries._id !== byProvider._id) { + await this.collection.deleteOne({ + _id: bySeries._id, + tenantId: master.tenantId, + principalId: master.principalId, + }); + } + const result = await this.collection.findOneAndUpdate( { + _id: byProvider._id, tenantId: master.tenantId, principalId: master.principalId, - "recurrence.kind": "exception", - "recurrence.seriesId": master._id, - "recurrence.recurrenceId": recurrenceId, }, { - // Mirror the master's ownership/calendar identity; set the - // instance's own content, schedule, provider identity, and cancelled - // flag. recurrence.kind/seriesId/recurrenceId are seeded from the - // filter on insert, so only cancelled is set here (setting the whole - // recurrence would conflict). $set: { - origin: master.origin, - calendarId: master.calendarId, - clientEventId: null, - connectionId: master.connectionId, - providerEventId, - providerVersion, - providerUpdatedAt: master.providerUpdatedAt, - deliveryState: master.connectionId - ? "confirmed" - : master.deliveryState, - providerMetadata: master.providerMetadata, - content: override.content, - schedule: override.schedule, - "recurrence.cancelled": override.cancelled, - lifecycleState: "active", - generation: master.generation, - updatedAt: now, - }, - $setOnInsert: { - _id: new ObjectId().toHexString() as EventId, - createdAt: now, - confirmedAt: now, + ...fields, + "recurrence.kind": "exception", + "recurrence.seriesId": master._id, + "recurrence.recurrenceId": recurrenceId, }, }, - { upsert: true, returnDocument: "after" }, + { returnDocument: "after" }, ); - if (!result) throw new Error("Exception upsert did not return a record"); + if (!result) { + throw new Error("Exception provider-identity update returned no record"); + } return EventRecordSchema.parse(result); } @@ -429,3 +613,12 @@ export class EventRepository { return result.deletedCount; } } + +function isDuplicateKeyError(error: unknown): boolean { + return ( + typeof error === "object" && + error !== null && + "code" in error && + (error as { code: unknown }).code === 11000 + ); +}