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

feat: support delta and full sync with the same code and simplify things

Browse Source 
Co-authored-by: Fredrik Adelöw <freben@gmail.com>
Co-authored-by: Johan Haals <johan.haals@gmail.com>
Signed-off-by: blam <ben@blam.sh>
This commit is contained in:
blam
2021-04-23 16:01:36 +02:00
parent 87ff2b455a
commit 3f6c50a529
2 changed files with 201 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.test.ts
+89
View File
@@ -294,5 +294,94 @@ describe('Default Processing Database', () => {
),
).toBeFalsy();
});
it('should add new locations using the delta options', async () => {
await processingDatabase!.transaction(async tx => {
await processingDatabase!.replaceUnprocessedEntities(tx, {
type: 'delta',
sourceKey: 'lols',
removed: [],
added: [
{
apiVersion: '1.0.0',
metadata: {
name: 'new-root',
},
kind: 'Location',
} as Entity,
],
});
});
const currentRefreshState = await db!<DbRefreshStateRow>(
'refresh_state',
).select();
const currentRefRowState = await db!<DbRefreshStateReferencesRow>(
'refresh_state_references',
).select();
expect(
currentRefreshState.some(
t => t.entity_ref === 'location:default/new-root',
),
).toBeTruthy();
expect(
currentRefRowState.some(
t =>
t.source_key === 'lols' &&
t.target_entity_ref === 'location:default/new-root',
),
).toBeTruthy();
});
it('should remove old locations using the delta options', async () => {
await createLocations(['location:default/new-root']);
await insertRefRow({
source_key: 'lols',
target_entity_ref: 'location:default/new-root',
});
await processingDatabase!.transaction(async tx => {
await processingDatabase!.replaceUnprocessedEntities(tx, {
type: 'delta',
sourceKey: 'lols',
added: [],
removed: [
{
apiVersion: '1.0.0',
metadata: {
name: 'new-root',
},
kind: 'Location',
} as Entity,
],
});
});
const currentRefreshState = await db!<DbRefreshStateRow>(
'refresh_state',
).select();
const currentRefRowState = await db!<DbRefreshStateReferencesRow>(
'refresh_state_references',
).select();
expect(
currentRefreshState.some(
t => t.entity_ref === 'location:default/new-root',
),
).toBeFalsy();
expect(
currentRefRowState.some(
t =>
t.source_key === 'lols' &&
t.target_entity_ref === 'location:default/new-root',
),
).toBeFalsy();
});
});
});
plugins/catalog-backend/src/next/database/DefaultProcessingDatabase.ts
+112 -130
View File
@@ -130,34 +130,48 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
);
}
private async createDelta(
tx: Knex.Transaction,
options: ReplaceUnprocessedEntitiesOptions,
): Promise<{ toAdd: Entity[]; toRemove: string[] }> {
if (options.type === 'delta') {
return {
toAdd: options.added,
toRemove: options.removed.map(e => stringifyEntityRef(e)),
};
}
const oldRefs = await tx<DbRefreshStateReferencesRow>(
'refresh_state_references',
)
.where({ source_key: options.sourceKey })
.select('target_entity_ref')
.then(rows => rows.map(r => r.target_entity_ref));
const items = options.items.map(entity => ({
entity,
ref: stringifyEntityRef(entity),
}));
const oldRefsSet = new Set(oldRefs);
const newRefsSet = new Set(items.map(item => item.ref));
const toAdd = items.filter(item => !oldRefsSet.has(item.ref));
const toRemove = oldRefs.filter(ref => !newRefsSet.has(ref));
return { toAdd: toAdd.map(({ entity }) => entity), toRemove };
}
async replaceUnprocessedEntities(
txOpaque: Transaction,
options: ReplaceUnprocessedEntitiesOptions,
): Promise<void> {
const tx = txOpaque as Knex.Transaction;
if (options.type === 'full') {
const oldRefs = await tx<DbRefreshStateReferencesRow>(
'refresh_state_references',
)
.where({ source_key: options.sourceKey })
.select('target_entity_ref')
.then(rows => rows.map(r => r.target_entity_ref));
const { toAdd, toRemove } = await this.createDelta(tx, options);
const items = options.items.map(entity => ({
entity,
ref: stringifyEntityRef(entity),
id: uuid(),
}));
const oldRefsSet = new Set(oldRefs);
const newRefsSet = new Set(items.map(item => item.ref));
const toAdd = items.filter(item => !oldRefsSet.has(item.ref));
const toRemove = oldRefs.filter(ref => !newRefsSet.has(ref));
if (toRemove.length) {
// TODO(freben): Batch split, to not hit variable limits?
/*
if (toRemove.length) {
// TODO(freben): Batch split, to not hit variable limits?
/*
WITH RECURSIVE
-- All the refs that can be reached from each individual root
root_reach(id, entity_ref) AS (
@@ -188,119 +202,87 @@ export class DefaultProcessingDatabase implements ProcessingDatabase {
AND refresh_state_references.target_entity_ref = "A"
AND them.id IS NULL;
*/
const removedCount = await tx<DbRefreshStateRow>('refresh_state')
.whereIn('entity_ref', function orphanedEntityRefs(orphans) {
return (
orphans
// All the refs that can be reached from each individual root
.withRecursive('root_reach', function rootReach(outer) {
// Start with all roots
return outer
.select({ id: 'id', entity_ref: 'target_entity_ref' })
.from('refresh_state_references')
.whereNotNull('source_key')
.union(function recurse(inner) {
return (
inner
// For each match, select all children
.select({
id: 'root_reach.id',
entity_ref:
'refresh_state_references.target_entity_ref',
})
.from('refresh_state_references')
.crossJoin('root_reach', {
'root_reach.entity_ref':
'refresh_state_references.source_entity_ref',
})
);
});
})
.select('us.entity_ref')
// Start out with our own matching row
.from('refresh_state_references')
.where('source_key', options.sourceKey)
.whereIn('target_entity_ref', toRemove)
// Expand the entire tree that emanates from that row
.leftJoin({ us: 'root_reach' }, function us() {
this.on('us.id', '=', 'refresh_state_references.id');
})
// Expand with all roots that target the same node but aren't ourselves
.leftOuterJoin({ them: 'root_reach' }, function them() {
this.on('them.entity_ref', '=', 'us.entity_ref');
this.andOn('them.id', '!=', 'us.id');
})
// Keep only the matches that had no other rooots
.whereNull('them.id')
);
})
.delete();
const removedCount = await tx<DbRefreshStateRow>('refresh_state')
.whereIn('entity_ref', function orphanedEntityRefs(orphans) {
return (
orphans
// All the refs that can be reached from each individual root
.withRecursive('root_reach', function rootReach(outer) {
// Start with all roots
return outer
.select({ id: 'id', entity_ref: 'target_entity_ref' })
.from('refresh_state_references')
.whereNotNull('source_key')
.union(function recurse(inner) {
return (
inner
// For each match, select all children
.select({
id: 'root_reach.id',
entity_ref:
'refresh_state_references.target_entity_ref',
})
.from('refresh_state_references')
.crossJoin('root_reach', {
'root_reach.entity_ref':
'refresh_state_references.source_entity_ref',
})
);
});
})
.select('us.entity_ref')
// Start out with our own matching row
.from('refresh_state_references')
.where('source_key', options.sourceKey)
.whereIn('target_entity_ref', toRemove)
// Expand the entire tree that emanates from that row
.leftJoin({ us: 'root_reach' }, function us() {
this.on('us.id', '=', 'refresh_state_references.id');
})
// Expand with all roots that target the same node but aren't ourselves
.leftOuterJoin({ them: 'root_reach' }, function them() {
this.on('them.entity_ref', '=', 'us.entity_ref');
this.andOn('them.id', '!=', 'us.id');
})
// Keep only the matches that had no other rooots
.whereNull('them.id')
);
})
.delete();
await tx<DbRefreshStateReferencesRow>('refresh_state_references')
.where('source_key', '=', options.sourceKey)
.whereIn('target_entity_ref', toRemove)
.delete();
await tx<DbRefreshStateReferencesRow>('refresh_state_references')
.where('source_key', '=', options.sourceKey)
.whereIn('target_entity_ref', toRemove)
.delete();
console.log(
`REMOVED, ${removedCount} entities: ${JSON.stringify(toRemove)}`,
);
}
this.logger.debug(
`removed, ${removedCount} entities: ${JSON.stringify(toRemove)}`,
);
}
if (toAdd.length) {
const state: Knex.DbRecord<DbRefreshStateRow>[] = toAdd.map(item => ({
entity_id: item.id,
entity_ref: item.ref,
unprocessed_entity: JSON.stringify(item.entity),
errors: '',
next_update_at: tx.fn.now(),
last_discovery_at: tx.fn.now(),
}));
const stateReferences: DbRefreshStateReferencesRow[] = toAdd.map(
item => ({
source_key: options.sourceKey,
target_entity_ref: item.ref,
}),
);
// TODO(freben): Concurrency? If we did these one by one, a .onConflict().merge would have made sense
await tx.batchInsert('refresh_state', state, BATCH_SIZE);
await tx.batchInsert(
'refresh_state_references',
stateReferences,
BATCH_SIZE,
);
}
if (toAdd.length) {
const state: Knex.DbRecord<DbRefreshStateRow>[] = toAdd.map(entity => ({
entity_id: uuid(),
entity_ref: stringifyEntityRef(entity),
unprocessed_entity: JSON.stringify(entity),
errors: '',
next_update_at: tx.fn.now(),
last_discovery_at: tx.fn.now(),
}));
// for (const entity of options.items) {
// const entityRef = stringifyEntityRef(entity);
// await tx<DbRefreshStateRow>('refresh_state')
// .insert({
// entity_id: uuid(),
// entity_ref: entityRef,
// unprocessed_entity: JSON.stringify(entity),
// errors: '',
// next_update_at: tx.fn.now(),
// last_discovery_at: tx.fn.now(),
// })
// .onConflict('entity_ref')
// .merge(['unprocessed_entity', 'last_discovery_at']);
// const [{ entity_id: entityId }] = await tx<DbRefreshStateRow>(
// 'refresh_state',
// ).where({ entity_ref: entityRef });
// entityIds.push(entityId);
// }
// const referenceRows: DbRefreshStateReferencesRow[] = entityIds.map(
// entityId => ({
// source_key: options.sourceKey,
// target_entityRef: entityRef,
// }),
// );
// await tx.batchInsert(
// 'refresh_state_references',
// referenceRows,
// BATCH_SIZE,
// );
// return;
const stateReferences: DbRefreshStateReferencesRow[] = toAdd.map(
entity => ({
source_key: options.sourceKey,
target_entity_ref: stringifyEntityRef(entity),
}),
);
// TODO(freben): Concurrency? If we did these one by one, a .onConflict().merge would have made sense
await tx.batchInsert('refresh_state', state, BATCH_SIZE);
await tx.batchInsert(
'refresh_state_references',
stateReferences,
BATCH_SIZE,
);
}
}