apps/backend/src/modules/external-integrations/workers/external-poll.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { createHash,randomUUID } from 'node:crypto';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { ExternalIntegrationService } from '../application/external-integration.service';
import { ExternalProviderRegistry } from '../providers/external-provider.registry';
@Injectable()
export class ExternalPollWorker {
constructor(
private readonly database:DatabaseService,
private readonly integrations:ExternalIntegrationService,
private readonly providers:ExternalProviderRegistry,
) {}
async pollConnection(input:{organizationId:string;locationId?:string;connectionId:string;streamKey:string}) {
const c=await this.integrations.load(input.organizationId,input.connectionId,input.locationId);
const provider=this.providers.get(c.provider);
if(!provider.capabilities.supportsPolling || !provider.poll) {
return {processed:false,reason:'polling_unsupported'};
}
const claimToken=randomUUID();
const claimed=await this.database.transaction(async(tx)=>tx.query<any>(
`insert into external_sync_cursors(
id,organization_id,integration_connection_id,stream_key,
poll_claim_token,poll_claimed_at
) values($1,$2,$3,$4,$5,now())
on conflict(integration_connection_id,stream_key)
do update set poll_claim_token=$5,poll_claimed_at=now(),updated_at=now()
where external_sync_cursors.poll_claimed_at is null
or external_sync_cursors.poll_claimed_at<now()-interval '15 minutes'
returning cursor_value`,
[randomUUID(),c.organization_id,c.id,input.streamKey,claimToken],
));
if(!claimed[0]) return {processed:false,reason:'poll_in_progress'};
let result;
try {
result=await provider.poll(this.integrations.context(c),{
streamKey:input.streamKey,cursor:claimed[0].cursor_value??undefined,
});
} catch(error) {
await this.releasePollClaim(c.id,input.streamKey,claimToken);
throw error;
}
try {
let inserted=0;
for(const event of result.events) {
const payload=JSON.stringify(event.payload??{});
const rows=await this.database.query<any>(
`insert into external_integration_inbox(
id,organization_id,integration_connection_id,provider,external_event_id,
event_type,external_entity_type,external_entity_id,payload_json,payload_hash,
provider_version,provider_updated_at
) values($1,$2,$3,$4,$5,$6,$7,$8,$9::jsonb,$10,$11,$12)
on conflict(integration_connection_id,external_event_id) do nothing
returning id`,
[
randomUUID(),c.organization_id,c.id,c.provider,event.externalEventId,event.eventType,
event.externalEntityType??null,event.externalEntityId??null,payload,
createHash('sha256').update(payload).digest('hex'),
event.providerVersion??null,event.providerUpdatedAt??null,
],
);
if(rows[0]) inserted++;
}
const cursorUpdated=await this.database.query(
`update external_sync_cursors
set cursor_value=$1,last_polled_at=now(),last_success_at=now(),
poll_claim_token=null,poll_claimed_at=null,updated_at=now()
where integration_connection_id=$2 and stream_key=$3 and poll_claim_token=$4
returning id`,
[result.cursor??claimed[0].cursor_value??null,c.id,input.streamKey,claimToken],
);
if(!cursorUpdated.length) return {processed:false,reason:'poll_claim_superseded'};
await this.database.query(
`update integration_connections set last_successful_sync_at=now() where id=$1`,
[c.id],
);
return {processed:true,inserted,cursor:result.cursor};
} catch(error) {
await this.releasePollClaim(c.id,input.streamKey,claimToken);
throw error;
}
}
private async releasePollClaim(connectionId:string,streamKey:string,claimToken:string) {
await this.database.query(
`update external_sync_cursors
set poll_claim_token=null,poll_claimed_at=null,updated_at=now()
where integration_connection_id=$1 and stream_key=$2 and poll_claim_token=$3`,
[connectionId,streamKey,claimToken],
);
}
}