paulononato/backstage
1
0
Fork 0
Code Issues Pull Requests Actions 45 Packages Projects Releases Wiki Activity

Catalog: Disallow insertion of conflicting entityRefs

Browse Source 
Co-authored-by: Patrik Oldsberg <poldsberg@gmail.com>

Signed-off-by: Johan Haals <johan.haals@gmail.com>
This commit is contained in:
Johan Haals
2021-06-28 14:02:57 +02:00
committed by Patrik Oldsberg
parent ffaedfb702
commit d0a13d227e
1 changed files with 67 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
+67 -26
View File
@@ -74,8 +74,14 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
location_key: locationKey,
})
.where('entity_id', id)
.andWhere('location_key', locationKey)
.orWhere('location_key', '');
.andWhere(inner => {
if (!locationKey) {
return inner.whereNull('location_key');
}
return inner
.where('location_key', locationKey)
.orWhereNull('location_key');
});
if (refreshResult === 0) {
throw new NotFoundError(`Processing state not found for ${id}`);
}
@@ -319,34 +325,69 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
): Promise<void> {
const tx = txOpaque as Knex.Transaction;
const stateRows = options.entities.map(
({ entity, locationKey }) =>
({
entity_id: uuid(),
entity_ref: stringifyEntityRef(entity),
unprocessed_entity: JSON.stringify(entity),
errors: '',
location_key: locationKey,
next_update_at: tx.fn.now(),
last_discovery_at: tx.fn.now(),
} as Knex.DbRecord<DbRefreshStateRow>),
);
const stateReferenceRows = stateRows.map(
stateRow =>
({
source_entity_ref: options.entityRef,
target_entity_ref: stateRow.entity_ref,
} as Knex.DbRecord<DbRefreshStateReferencesRow>),
);
const stateReferenceRows = new Array<
Knex.DbRecord<DbRefreshStateReferencesRow>
>();
// Upsert all of the unprocessed entities into the refresh_state table, by
// their entity ref.
// TODO(freben): Can this be batched somehow?
for (const row of stateRows) {
await tx<DbRefreshStateRow>('refresh_state')
.insert(row)
.onConflict('entity_ref')
.merge(['unprocessed_entity', 'last_discovery_at']);
for (const { entity, locationKey } of options.entities) {
const entityRef = stringifyEntityRef(entity);
const serializedEntity = JSON.stringify(entity);
const refreshResult = await tx<DbRefreshStateRow>('refresh_state')
.update({
unprocessed_entity: serializedEntity,
location_key: locationKey,
last_discovery_at: tx.fn.now(),
})
.where('entity_ref', entityRef)
.andWhere(inner => {
if (!locationKey) {
return inner.whereNull('location_key');
}
return inner
.where('location_key', locationKey)
.orWhereNull('location_key');
});
if (refreshResult === 0) {
const result = await tx<DbRefreshStateRow>('refresh_state')
.insert({
entity_id: uuid(),
entity_ref: entityRef,
unprocessed_entity: serializedEntity,
errors: '',
location_key: locationKey,
next_update_at: tx.fn.now(),
last_discovery_at: tx.fn.now(),
})
.onConflict('entity_ref')
.ignore();
if (result.length === 0) {
// Detected conflicting entity with same entityRef but different locationKey
const [conflictingEntity] = await tx<DbRefreshStateRow>(
'refresh_state',
)
.where({ entity_ref: entityRef })
.select();
if (conflictingEntity.location_key !== locationKey) {
this.options.logger.warn(
`Detected conflicting entityRef ${entityRef} already referenced by ${conflictingEntity.location_key} and now also ${locationKey}`,
);
continue;
}
}
}
// Skipped on locationKey conflict
stateReferenceRows.push({
source_entity_ref: options.entityRef,
target_entity_ref: entityRef,
});
}
// Replace all references for the originating entity before creating new ones