Skip to content

Commit 0140590

Browse files
committed
Update backends.service.ts
1 parent d38ce16 commit 0140590

1 file changed

Lines changed: 62 additions & 40 deletions

File tree

‎apps/api/src/core/backends/backends.service.ts‎

Lines changed: 62 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,7 @@ export class BackendsService extends AbstractQueueProcessor {
7575
if (isSyncedJob?.concernedTo?.$ref === 'groups') {
7676
await this.applyGroupJobResult(isSyncedJob.concernedTo.id, result?.jobName, true);
7777
} else {
78+
await this.markGroupsOfFirstSyncedMember(isSyncedJob?.concernedTo?.id);
7879
await this.identitiesService.model.findByIdAndUpdate(isSyncedJob?.concernedTo?.id, {
7980
$set: {
8081
state: IdentityState.SYNCED,
@@ -167,10 +168,14 @@ export class BackendsService extends AbstractQueueProcessor {
167168
this.logger.error(`Set State on error [${payload.jobId}]`);
168169
myState = IdentityState.ON_ERROR;
169170
}
171+
if (jState === JobState.COMPLETED && result.jobName !== ActionType.IDENTITY_DELETE) {
172+
await this.markGroupsOfFirstSyncedMember(completedJob?.concernedTo?.id);
173+
}
170174
await this.identitiesService.model.findByIdAndUpdate(completedJob?.concernedTo?.id, {
171175
$set: {
172176
state: myState,
173-
lastBackendSync: jState === JobState.COMPLETED ? new Date() : null,
177+
// en cas d'échec la date est conservée : l'identité existe toujours dans les backends
178+
...(jState === JobState.COMPLETED ? { lastBackendSync: new Date() } : {}),
174179
deletedFlag: result.jobName === ActionType.IDENTITY_DELETE,
175180
},
176181
});
@@ -559,6 +564,24 @@ export class BackendsService extends AbstractQueueProcessor {
559564
await this.identitiesService.model.findByIdAndUpdate(id, { $set: { state } });
560565
}
561566

567+
/**
568+
* À appeler juste avant d'enregistrer la synchronisation réussie d'une identité : si elle n'avait encore jamais
569+
* été synchronisée, elle était absente du payload de ses groupes, qui repassent à TO_SYNC pour l'inclure.
570+
* Fait avant la mise à jour de l'identité pour que syncAll trouve ces groupes en attente.
571+
*/
572+
protected async markGroupsOfFirstSyncedMember(identityId?: Types.ObjectId | string): Promise<void> {
573+
if (!identityId) return;
574+
const identity = await this.identitiesService.model
575+
.findById(identityId, { lastBackendSync: 1 })
576+
.lean<Identities>()
577+
.exec();
578+
if (!identity || identity.lastBackendSync) return;
579+
580+
await this.groupsService.model
581+
.updateMany({ member: identity._id, type: { $ne: GroupType.SUPER } }, { $set: { state: IdentityState.TO_SYNC } })
582+
.exec();
583+
}
584+
562585
/**
563586
* Applique le résultat d'un job GROUP_* sur le groupe concerné
564587
*/
@@ -585,12 +608,19 @@ export class BackendsService extends AbstractQueueProcessor {
585608

586609
/**
587610
* Construit le payload envoyé au daemon pour un groupe : le groupe et ses membres résolus
588-
* (hors identités supprimées et DONT_SYNC, qui ne peuvent pas être membres)
611+
* (hors identités supprimées, DONT_SYNC et jamais synchronisées)
589612
*/
590613
protected async buildGroupPayload(group: Groups): Promise<Record<string, any>> {
591614
const members = await this.identitiesService.model
592615
.find(
593-
{ _id: { $in: group.member || [] }, deletedFlag: { $ne: true }, state: { $ne: IdentityState.DONT_SYNC } },
616+
{
617+
_id: { $in: group.member || [] },
618+
deletedFlag: { $ne: true },
619+
state: { $ne: IdentityState.DONT_SYNC },
620+
// une identité jamais synchronisée n'existe pas encore dans les backends : le groupe est envoyé sans elle
621+
// et repassé à TO_SYNC lors de sa première synchronisation (markGroupsOfFirstSyncedMember)
622+
lastBackendSync: { $ne: null },
623+
},
594624
{
595625
'inetOrgPerson.cn': 1,
596626
'inetOrgPerson.uid': 1,
@@ -641,11 +671,14 @@ export class BackendsService extends AbstractQueueProcessor {
641671
if (!jobOptions.async) return { ...identities, ...(await this.syncAllGroups(jobOptions)) };
642672

643673
const syncGroups = async () => {
674+
const deadline = Date.now() + IDENTITIES_BEFORE_GROUPS_TIMEOUT_MS;
644675
await this.waitForJobs(
645676
Object.values(identities)
646677
.map((job: Partial<Jobs>) => job?.jobId)
647678
.filter(Boolean),
679+
deadline,
648680
);
681+
await this.waitForIdentitiesProcessed(Object.keys(identities), deadline);
649682
return await this.syncAllGroups(jobOptions);
650683
};
651684
if (waitForGroups) return { ...identities, ...(await syncGroups()) };
@@ -657,10 +690,9 @@ export class BackendsService extends AbstractQueueProcessor {
657690
}
658691

659692
/**
660-
* Attend la fin (succès ou échec) des jobs donnés, dans la limite de IDENTITIES_BEFORE_GROUPS_TIMEOUT_MS au total
693+
* Attend la fin (succès ou échec) des jobs donnés, au plus tard jusqu'à `deadline`
661694
*/
662-
protected async waitForJobs(jobIds: string[]): Promise<void> {
663-
const deadline = Date.now() + IDENTITIES_BEFORE_GROUPS_TIMEOUT_MS;
695+
protected async waitForJobs(jobIds: string[], deadline: number): Promise<void> {
664696
for (const jobId of jobIds) {
665697
const remaining = deadline - Date.now();
666698
if (remaining <= 0) {
@@ -675,6 +707,26 @@ export class BackendsService extends AbstractQueueProcessor {
675707
}
676708
}
677709

710+
/**
711+
* La fin d'un job est notifiée avant que l'écouteur « completed » de l'API ait mis à jour l'identité
712+
* (état, lastBackendSync) : on attend que les identités envoyées ne soient plus à TO_SYNC / PROCESSING,
713+
* sinon leurs groupes seraient considérés comme ayant des membres non synchronisés
714+
*/
715+
protected async waitForIdentitiesProcessed(ids: string[], deadline: number): Promise<void> {
716+
if (!ids.length) return;
717+
const filter = {
718+
_id: { $in: ids.filter((id) => Types.ObjectId.isValid(id)) },
719+
state: { $in: [IdentityState.TO_SYNC, IdentityState.PROCESSING] },
720+
};
721+
while ((await this.identitiesService.model.countDocuments(filter).exec()) > 0) {
722+
if (Date.now() >= deadline) {
723+
this.logger.warn('Identities still pending after their jobs finished, syncing groups anyway');
724+
return;
725+
}
726+
await new Promise((resolve) => setTimeout(resolve, 1_000));
727+
}
728+
}
729+
678730
/**
679731
* Synchronise tous les groupes à l'état TO_SYNC (appelé par syncAll une fois les identités synchronisées,
680732
* pour que les membres existent dans les backends avant leurs groupes)
@@ -704,43 +756,12 @@ export class BackendsService extends AbstractQueueProcessor {
704756
}
705757
if (!groups.length) return {};
706758

707-
// un groupe n'est synchronisé que si tous ses membres (hors identités supprimées) le sont déjà dans les backends ;
708-
// sinon il reste à TO_SYNC et sera repris par une prochaine synchronisation
709-
const unsyncedMembers = new Set(
710-
(
711-
await this.identitiesService.model
712-
.find(
713-
{
714-
_id: { $in: groups.flatMap((group) => group.member || []) },
715-
deletedFlag: { $ne: true },
716-
// les identités DONT_SYNC ne sont pas envoyées comme membres, elles ne bloquent pas le groupe
717-
state: { $nin: [IdentityState.SYNCED, IdentityState.DONT_SYNC] },
718-
},
719-
{ _id: 1 },
720-
)
721-
.lean()
722-
.exec()
723-
).map((identity) => `${identity._id}`),
724-
);
725-
726-
const result = {};
727-
const ready: Groups[] = [];
728-
for (const group of groups) {
729-
const pending = (group.member || []).filter((id) => unsyncedMembers.has(`${id}`)).length;
730-
if (!pending) {
731-
ready.push(group);
732-
continue;
733-
}
734-
this.logger.warn(`Group ${group.cn} not synced: ${pending} member(s) not synced yet`);
735-
result[`${group._id}`] = { error: `${pending} membre(s) non synchronisé(s), groupe en attente` };
736-
}
737-
if (!ready.length) return result;
738-
739759
const task: Document<Tasks> = await this.tasksService.create<Tasks>({
740-
jobs: ready.map((group) => group._id),
760+
jobs: groups.map((group) => group._id),
741761
});
742762

743-
for (const group of ready) {
763+
const result = {};
764+
for (const group of groups) {
744765
const action = group.lastBackendSync ? ActionType.GROUP_UPDATE : ActionType.GROUP_CREATE;
745766
try {
746767
const [executedJob] = await this.executeJob(action, group._id, await this.buildGroupPayload(group), {
@@ -909,6 +930,7 @@ export class BackendsService extends AbstractQueueProcessor {
909930
if (concernedTo && !!options?.updateStatus && isGroupJob) {
910931
await this.applyGroupJobResult(concernedTo, actionType, true);
911932
} else if (concernedTo && !!options?.updateStatus) {
933+
if (options?.dataState !== DataStatusEnum.DELETED) await this.markGroupsOfFirstSyncedMember(concernedTo);
912934
await this.identitiesService.model.findByIdAndUpdate(concernedTo, {
913935
$set: {
914936
state: options?.targetState || IdentityState.SYNCED,

0 commit comments

Comments
 (0)