@@ -2,6 +2,7 @@ import { db } from '@sim/db'
22import { peopleDirectoryScimChange as change , peopleDirectorySync as sync } from '@sim/db/schema'
33import { createLogger } from '@sim/logger'
44import { getPostgresErrorCode } from '@sim/utils/errors'
5+ import { chunkArray } from '@sim/utils/helpers'
56import { generateId } from '@sim/utils/id'
67import { backoffWithJitter } from '@sim/utils/retry'
78import { and , eq , isNotNull , isNull , lte , or , sql } from 'drizzle-orm'
@@ -40,7 +41,7 @@ export const CLEARED_RUN = {
4041 cursor : null ,
4142 usersRead : 0 ,
4243} as const
43- const RELEASED = { leaseToken : null , leaseExpiresAt : null } as const
44+ export const RELEASED = { leaseToken : null , leaseExpiresAt : null } as const
4445const available = ( at : Date ) => or ( isNull ( sync . leaseToken ) , lte ( sync . leaseExpiresAt , at ) )
4546const due = ( at : Date ) => or ( lte ( sync . nextSyncAt , at ) , isNotNull ( pendingScimChange ) )
4647type Connection = typeof sync . $inferSelect
@@ -78,11 +79,11 @@ const ownsLease = (worker: Worker) =>
7879/** Every page and its next cursor commit together, after checking that the same run still owns the source. */
7980async function checkpoint (
8081 worker : Worker ,
81- step : 'page' | 'publish' ,
82- write : ( tx : DbTransaction , row : Checkpoint ) => Promise < void >
82+ write : ( tx : DbTransaction , row : Checkpoint ) => Promise < void > ,
83+ options : { lockProfiles ?: boolean } = { }
8384) {
8485 return db . transaction ( async ( tx ) => {
85- if ( step === 'publish' ) await lockProfiles ( tx , worker . organizationId )
86+ if ( options . lockProfiles ) await lockProfiles ( tx , worker . organizationId )
8687 const [ row ] = await tx
8788 . select ( checkpointColumns )
8889 . from ( sync )
@@ -127,7 +128,7 @@ function retryUpdate(row: Checkpoint, providerDelay: number, reason: string) {
127128
128129async function checkpointUsers ( worker : Worker , reader : DirectoryReader , cursor : string | null ) {
129130 const page = await reader . users ( cursor )
130- return checkpoint ( worker , 'page' , async ( tx ) => {
131+ return checkpoint ( worker , async ( tx ) => {
131132 await stageUsers ( tx , worker , cursor , page . people )
132133 await tx
133134 . update ( sync )
@@ -147,7 +148,7 @@ async function checkpointReports(worker: Worker, reader: DirectoryReader) {
147148 throw new OrchestrationError ( 'validation' , 'The directory cannot read reporting lines.' )
148149 const items = await pendingManagers ( db , worker )
149150 if ( ! items . length ) {
150- return checkpoint ( worker , 'page' , async ( tx ) => {
151+ return checkpoint ( worker , async ( tx ) => {
151152 await tx . update ( sync ) . set ( { phase : 'publish' } ) . where ( ownsLease ( worker ) )
152153 } )
153154 }
@@ -168,7 +169,7 @@ async function checkpointReports(worker: Worker, reader: DirectoryReader) {
168169 const delays = results . flatMap ( ( result ) =>
169170 'retryAfterMs' in result ? [ result . retryAfterMs ] : [ ]
170171 )
171- const applied = await checkpoint ( worker , 'page' , async ( tx , row ) => {
172+ const applied = await checkpoint ( worker , async ( tx , row ) => {
172173 await stageRelationships ( tx , worker , pages )
173174 const update = delays . length
174175 ? retryUpdate (
@@ -184,12 +185,16 @@ async function checkpointReports(worker: Worker, reader: DirectoryReader) {
184185
185186/** A local SCIM generation captures, validates, and publishes atomically. New provisioning changes have their own marker. */
186187async function runScim ( worker : Worker ) {
187- await checkpoint ( worker , 'publish' , async ( tx , row ) => {
188- await discardStage ( tx , row . organizationId )
189- const count = await stageScimDirectory ( tx , row . organizationId , worker . runId , row . sourceTarget )
190- await tx . update ( sync ) . set ( { phase : 'publish' , usersRead : count } ) . where ( ownsLease ( worker ) )
191- await publish ( tx , worker , 'scim' )
192- } )
188+ await checkpoint (
189+ worker ,
190+ async ( tx , row ) => {
191+ await discardStage ( tx , row . organizationId )
192+ const count = await stageScimDirectory ( tx , row . organizationId , worker . runId , row . sourceTarget )
193+ await tx . update ( sync ) . set ( { phase : 'publish' , usersRead : count } ) . where ( ownsLease ( worker ) )
194+ await publish ( tx , worker , 'scim' )
195+ } ,
196+ { lockProfiles : true }
197+ )
193198}
194199
195200async function handleFailure ( worker : Worker , error : unknown ) {
@@ -203,7 +208,7 @@ async function handleFailure(worker: Worker, error: unknown) {
203208 retry || ( error instanceof OrchestrationError && error . code !== 'internal' )
204209 ? error . message
205210 : 'Directory sync was interrupted. Sim will retry automatically.'
206- const applied = await checkpoint ( worker , 'page' , async ( tx , row ) => {
211+ const applied = await checkpoint ( worker , async ( tx , row ) => {
207212 if ( temporary ) {
208213 await tx
209214 . update ( sync )
@@ -297,7 +302,7 @@ export async function runDirectorySync(organizationId: string, trigger: keyof ty
297302 if ( ! current ) return true
298303 await requirePeopleEnabled ( organizationId )
299304 if ( current . phase === 'publish' ) {
300- await checkpoint ( worker , 'publish' , ( tx ) => publish ( tx , worker , row . source ) )
305+ await checkpoint ( worker , ( tx ) => publish ( tx , worker , row . source ) , { lockProfiles : true } )
301306 return true
302307 }
303308 reader ??= await connectReader ( row )
@@ -340,8 +345,8 @@ export async function syncDueDirectories() {
340345 SELECT organization_id AS "organizationId" FROM candidates
341346 GROUP BY organization_id ORDER BY min(due_at), organization_id LIMIT 50
342347 ` )
343- for ( let i = 0 ; i < connections . length && Date . now ( ) < deadline ; i += 2 ) {
344- const slice = connections . slice ( i , i + 2 )
348+ for ( const slice of chunkArray ( connections , 2 ) ) {
349+ if ( Date . now ( ) >= deadline ) break
345350 const results = await Promise . allSettled (
346351 slice . map ( ( { organizationId } ) => runDirectorySync ( organizationId , 'cron' ) )
347352 )
0 commit comments