apps/backend/src/modules/commercial/workers/scheduled-plan-change.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { randomUUID } from 'node:crypto';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { SaaSBillingProviderRegistry } from '../providers/saas-billing-provider.registry';
import { EntitlementProjectionService } from '../application/entitlement-projection.service';
@Injectable()
export class ScheduledPlanChangeWorker {
constructor(
private readonly database:DatabaseService,
private readonly providers:SaaSBillingProviderRegistry,
private readonly entitlements:EntitlementProjectionService,
) {}
async runBatch(limit=50) {
const claimToken=randomUUID();
const rows=await this.database.query<any>(
`with claim as (
select s.id
from organization_subscriptions s
where s.scheduled_plan_version_id is not null
and s.scheduled_change_at is not null and s.scheduled_change_at<=now()
and s.status in ('active','past_due','grace_period')
and (s.scheduled_change_claimed_at is null
or s.scheduled_change_claimed_at<now()-interval '15 minutes')
order by s.scheduled_change_at
limit $1
for update skip locked
), claimed as (
update organization_subscriptions s
set scheduled_change_claim_token=$2,scheduled_change_claimed_at=now(),updated_at=now()
from claim
where s.id=claim.id
returning s.*
)
select claimed.*,ba.provider,ba.provider_customer_id,ba.id as billing_account_pk,
pv.provider_price_id as target_provider_price_id
from claimed
join billing_accounts ba on ba.id=claimed.billing_account_id
join saas_plan_versions pv on pv.id=claimed.scheduled_plan_version_id`,
[limit,claimToken],
);
let completed=0;
for(const row of rows) {
try {
const provider=this.providers.get(row.provider);
const result=await provider.changeSubscription({
organizationId:row.organization_id,billingAccountId:row.billing_account_pk,
providerCustomerId:row.provider_customer_id,
credentialReference:String(process.env.SAAS_BILLING_CREDENTIAL_REFERENCE??'env:STRIPE_SECRET_KEY'),
config:{},
},{
providerSubscriptionId:row.provider_subscription_id,
providerPriceId:row.target_provider_price_id,
effective:'immediate',
idempotencyKey:`scheduled-plan-change:${row.id}:${row.scheduled_plan_version_id}`,
});
const applied=await this.database.transaction(async(tx)=>{
const updated=await tx.query(
`update organization_subscriptions
set plan_version_id=scheduled_plan_version_id,provider_price_id=$1,
provider_status=$2,scheduled_plan_version_id=null,scheduled_change_at=null,
scheduled_change_claim_token=null,scheduled_change_claimed_at=null,
updated_at=now(),version=version+1
where id=$3 and scheduled_change_claim_token=$4
returning id`,
[row.target_provider_price_id,result.providerStatus,row.id,claimToken],
);
if(!updated[0]) return false;
await tx.query(
`update subscription_change_requests
set status='completed',completed_at=now()
where subscription_id=$1 and to_plan_version_id=$2 and status='scheduled'`,
[row.id,row.scheduled_plan_version_id],
);
return true;
});
if(!applied) continue;
await this.entitlements.rebuild(row.organization_id);
completed++;
} catch {
await this.database.query(
`update organization_subscriptions
set scheduled_change_claim_token=null,scheduled_change_claimed_at=null,updated_at=now()
where id=$1 and scheduled_change_claim_token=$2`,
[row.id,claimToken],
);
await this.database.query(
`update subscription_change_requests
set status='failed'
where subscription_id=$1 and to_plan_version_id=$2 and status='scheduled'`,
[row.id,row.scheduled_plan_version_id],
);
}
}
return {processed:rows.length,completed};
}
}