29496424463

Committed 16 Jul 2026 11:59AM UTC coverage: 58.911% (-0.05%) from 58.958%

Build # 29496424463

Build Type

Pull #3612

github

Committed by

web-flow

Commit Message

Merge aed64ec4f into 52c387e49

Pull Request Pull Request #3612: [sync] test: mock prisma setting findMany response in InvitationService tests

Coverage Stats

8097 of 10316 branches covered (78.49%)

17 of 47 new or added lines in 6 files covered. (36.17%)

17 existing lines in 4 files now uncovered.

35287 of 59899 relevant lines covered (58.91%)

3006.12 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

44.94

/apps/nestjs-backend/src/features/base/base-import-processor/base-import-junction.processor.ts

/* eslint-disable @typescript-eslint/naming-convention */
import { InjectQueue, Processor, WorkerHost } from '@nestjs/bullmq';
import { Injectable, Logger } from '@nestjs/common';
import {
  PrismaClientKnownRequestError,
  PrismaClientUnknownRequestError,
} from '@prisma/client/runtime/library';
import type { ILinkFieldOptions } from '@teable/core';
import { FieldType } from '@teable/core';
import { PrismaService } from '@teable/db-main-prisma';
import type { IBaseJson } from '@teable/openapi';
import { UploadType } from '@teable/openapi';
import type { Job } from 'bullmq';
import { Queue } from 'bullmq';
import * as csvParser from 'csv-parser';
import * as unzipper from 'unzipper';
import { InjectDbProvider } from '../../../db-provider/db.provider';
import { IDbProvider } from '../../../db-provider/db.provider.interface';
import { DataDbClientManager } from '../../../global/data-db-client-manager.service';
import StorageAdapter from '../../attachments/plugins/adapter';
import { InjectStorageAdapter } from '../../attachments/plugins/storage';
import { createFieldInstanceByRaw } from '../../field/model/factory';
import { PersistedComputedBackfillService } from '../../record/computed/services/persisted-computed-backfill.service';
import { BatchProcessor } from '../BatchProcessor.class';

type IDataPrismaExecutor = {
  $queryRawUnsafe<T = unknown>(query: string, ...values: unknown[]): Promise<T>;
  $executeRawUnsafe(query: string, ...values: unknown[]): Promise<number>;
};

type IDataPrismaScopedClient = IDataPrismaExecutor & {
  $tx?: <T>(fn: (prisma: IDataPrismaExecutor) => Promise<T>) => Promise<T>;
  $transaction?: <T>(fn: (prisma: IDataPrismaExecutor) => Promise<T>) => Promise<T>;
};

interface IBaseImportJunctionCsvJob {
  path: string;
  baseId: string;
  tableIdMap: Record<string, string>;
  fieldIdMap: Record<string, string>;
  structure: IBaseJson;
}

export const BASE_IMPORT_JUNCTION_CSV_QUEUE = 'base-import-junction-csv-queue';

@Injectable()
@Processor(BASE_IMPORT_JUNCTION_CSV_QUEUE)
export class BaseImportJunctionCsvQueueProcessor extends WorkerHost {
  private logger = new Logger(BaseImportJunctionCsvQueueProcessor.name);
  private processedJobs = new Set<string>();

  constructor(
    private readonly prismaService: PrismaService,
    private readonly persistedComputedBackfillService: PersistedComputedBackfillService,
    @InjectStorageAdapter() private readonly storageAdapter: StorageAdapter,
    @InjectQueue(BASE_IMPORT_JUNCTION_CSV_QUEUE)
    public readonly queue: Queue<IBaseImportJunctionCsvJob>,
    @InjectDbProvider() private readonly dbProvider: IDbProvider,
    private readonly dataDbClientManager: DataDbClientManager
  ) {
    super();
  }

  public async process(job: Job<IBaseImportJunctionCsvJob>) {
    const jobId = String(job.id);
    if (this.processedJobs.has(jobId)) {
      this.logger.log(`Job ${jobId} already processed, skipping`);
      return;
    }

    this.processedJobs.add(jobId);

    const { path, baseId, tableIdMap, fieldIdMap, structure } = job.data;

    try {
      await this.importJunctionChunk(path, baseId, fieldIdMap, structure);
      await this.persistedComputedBackfillService.recomputeForTables(Object.values(tableIdMap));
    } catch (error) {
      this.logger.error(
        `Process base import junction csv failed: ${(error as Error)?.message}`,
        (error as Error)?.stack
      );
    }
  }

  private async importJunctionChunk(
    path: string,
    baseId: string,
    fieldIdMap: Record<string, string>,
    structure: IBaseJson
  ) {
    const csvStream = await this.storageAdapter.downloadFile(
      StorageAdapter.getBucket(UploadType.Import),
      path
    );

    const sourceLinkFields = structure.tables
      .map(({ fields }) => fields)
      .flat()
      .filter((f) => f.type === FieldType.Link && !f.isLookup);

    const linkFieldRaws = await this.prismaService.field.findMany({
      where: {
        id: {
          in: Object.values(fieldIdMap),
        },
        type: FieldType.Link,
        isLookup: null,
      },
    });

    const junctionDbTableNameMap = {} as Record<
      string,
      {
        sourceSelfKeyName: string;
        sourceForeignKeyName: string;
        targetSelfKeyName: string;
        targetForeignKeyName: string;
        targetFkHostTableName: string;
      }
    >;

    const linkFieldInstances = linkFieldRaws.map((f) => createFieldInstanceByRaw(f));

    for (const sourceField of sourceLinkFields) {
      const { options: sourceOptions } = sourceField;
      const {
        fkHostTableName: sourceFkHostTableName,
        selfKeyName: sourceSelfKeyName,
        foreignKeyName: sourceForeignKeyName,
      } = sourceOptions as ILinkFieldOptions;
      const targetField = linkFieldInstances.find((f) => f.id === fieldIdMap[sourceField.id])!;
      const { options: targetOptions } = targetField;
      const {
        fkHostTableName: targetFkHostTableName,
        selfKeyName: targetSelfKeyName,
        foreignKeyName: targetForeignKeyName,
      } = targetOptions as ILinkFieldOptions;
      if (sourceFkHostTableName.includes('junction_')) {
        junctionDbTableNameMap[sourceFkHostTableName] = {
          sourceSelfKeyName,
          sourceForeignKeyName,
          targetSelfKeyName,
          targetForeignKeyName,
          targetFkHostTableName,
        };
      }
    }

    const parser = unzipper.Parse();
    csvStream.pipe(parser);

    const processedFiles = new Set<string>();

    return new Promise<{ success: boolean }>((resolve, reject) => {
      parser.on('entry', (entry) => {
        const filePath = entry.path;

        if (processedFiles.has(filePath)) {
          entry.autodrain();
          return;
        }
        processedFiles.add(filePath);

        if (
          filePath.startsWith('tables/') &&
          entry.type !== 'Directory' &&
          filePath.includes('junction_')
        ) {
          const name = filePath.replace('tables/', '').split('.');
          name.pop();
          const junctionTableName = name.join('.');
          const junctionInfo = junctionDbTableNameMap[junctionTableName];

          const {
            sourceForeignKeyName,
            targetForeignKeyName,
            sourceSelfKeyName,
            targetSelfKeyName,
            targetFkHostTableName,
          } = junctionInfo;

          const batchProcessor = new BatchProcessor<Record<string, unknown>>((chunk) =>
            this.handleJunctionChunk(baseId, chunk, targetFkHostTableName)
          );

          entry
            .pipe(
              csvParser.default({
                // strict: true,
                mapValues: ({ value }) => {
                  // deal with old junction order case
                  return value === '' ? null : value;
                },
                mapHeaders: ({ header }) => {
                  return header
                    .replaceAll(sourceForeignKeyName, targetForeignKeyName)
                    .replaceAll(sourceSelfKeyName, targetSelfKeyName);
                },
              })
            )
            .pipe(batchProcessor)
            .on('error', (error: Error) => {
              this.logger.error(`process csv import error: ${error.message}`, error.stack);
              reject(error);
            })
            .on('end', () => {
              this.logger.log(`csv ${junctionTableName} finished`);
            });
        } else {
          entry.autodrain();
        }
      });

      parser.on('close', () => {
        this.logger.log('import csv junction completed');
        resolve({ success: true });
      });

      parser.on('error', (error) => {
        this.logger.error(`import csv junction parser error: ${error.message}`, error.stack);
        reject(error);
      });
    });
  }

  private async dataTransaction<T>(
    dataPrisma: IDataPrismaScopedClient,
    fn: (prisma: IDataPrismaExecutor) => Promise<T>
  ) {
    if (dataPrisma.$tx) {
      return await dataPrisma.$tx(fn);
    }

    if (dataPrisma.$transaction) {
      return await dataPrisma.$transaction(fn);
    }

    return await fn(dataPrisma);
  }

  private async handleJunctionChunk(
    baseId: string,
    results: Record<string, unknown>[],
    targetFkHostTableName: string
  ) {
    const allForeignKeyInfos = [] as {
      constraint_name: string;
      column_name: string;
      referenced_table_schema: string;
      referenced_table_name: string;
      referenced_column_name: string;
      dbTableName: string;
    }[];

    const dataPrisma = (await this.dataDbClientManager.dataPrismaForBase(
      baseId
    )) as IDataPrismaScopedClient;
    const dataKnex = await this.dataDbClientManager.dataKnexForBase(baseId);

    await this.dataTransaction(dataPrisma, async (prisma) => {
      // delete foreign keys if(exist) then duplicate table data
      const foreignKeysInfoSql = this.dbProvider.getForeignKeysInfo(targetFkHostTableName);
      const foreignKeysInfo = await prisma.$queryRawUnsafe<
        {
          constraint_name: string;
          column_name: string;
          referenced_table_schema: string;
          referenced_table_name: string;
          referenced_column_name: string;
        }[]
      >(foreignKeysInfoSql);
      const newForeignKeyInfos = foreignKeysInfo.map((info) => ({
        ...info,
        dbTableName: targetFkHostTableName,
      }));
      allForeignKeyInfos.push(...newForeignKeyInfos);

      for (const { constraint_name, column_name, dbTableName } of allForeignKeyInfos) {
        const dropForeignKeyQuery = dataKnex.schema
          .alterTable(dbTableName, (table) => {
            table.dropForeign(column_name, constraint_name);
          })
          .toQuery();

        await prisma.$executeRawUnsafe(dropForeignKeyQuery);
      }

      const sql = dataKnex.table(targetFkHostTableName).insert(results).toQuery();
      try {
        await prisma.$executeRawUnsafe(sql);
      } catch (error) {
        if (error instanceof PrismaClientKnownRequestError) {
          this.logger.error(
            `exc junction import task known error: (${error.code}): ${error.message}`,
            error.stack
          );
        } else if (error instanceof PrismaClientUnknownRequestError) {
          this.logger.error(
            `exc junction import task unknown error: ${error.message}`,
            error.stack
          );
        } else {
          this.logger.error(
            `exc junction import task error: ${(error as Error)?.message}`,
            (error as Error)?.stack
          );
        }
      }

      // add foreign keys with NOT VALID to skip existing data validation
      for (const {
        constraint_name,
        column_name,
        dbTableName,
        referenced_table_schema: referencedTableSchema,
        referenced_table_name: referencedTableName,
        referenced_column_name: referencedColumnName,
      } of allForeignKeyInfos) {
        const [schema, tableName] = dbTableName.split('.');
        const addForeignKeyQuery = dataKnex
          .raw(
            'ALTER TABLE ??.?? ADD CONSTRAINT ?? FOREIGN KEY (??) REFERENCES ??.??(??) NOT VALID',
            [
              schema,
              tableName,
              constraint_name,
              column_name,
              referencedTableSchema,
              referencedTableName,
              referencedColumnName,
            ]
          )
          .toQuery();
        await prisma.$executeRawUnsafe(addForeignKeyQuery);
      }
    });
  }
}

1	/* eslint-disable @typescript-eslint/naming-convention */
2	import { InjectQueue, Processor, WorkerHost } from '@nestjs/bullmq';
3	import { Injectable, Logger } from '@nestjs/common';
4	import {
5	PrismaClientKnownRequestError,
6	PrismaClientUnknownRequestError,
7	} from '@prisma/client/runtime/library';
8	import type { ILinkFieldOptions } from '@teable/core';
9	import { FieldType } from '@teable/core';
10	import { PrismaService } from '@teable/db-main-prisma';
11	import type { IBaseJson } from '@teable/openapi';
12	import { UploadType } from '@teable/openapi';
13	import type { Job } from 'bullmq';
14	import { Queue } from 'bullmq';
15	import * as csvParser from 'csv-parser';
16	import * as unzipper from 'unzipper';
17	import { InjectDbProvider } from '../../../db-provider/db.provider';
18	import { IDbProvider } from '../../../db-provider/db.provider.interface';
19	import { DataDbClientManager } from '../../../global/data-db-client-manager.service';
20	import StorageAdapter from '../../attachments/plugins/adapter';
21	import { InjectStorageAdapter } from '../../attachments/plugins/storage';
22	import { createFieldInstanceByRaw } from '../../field/model/factory';
23	import { PersistedComputedBackfillService } from '../../record/computed/services/persisted-computed-backfill.service';
24	import { BatchProcessor } from '../BatchProcessor.class';
25
26	type IDataPrismaExecutor = {
27	$queryRawUnsafe<T = unknown>(query: string, ...values: unknown[]): Promise<T>;
28	$executeRawUnsafe(query: string, ...values: unknown[]): Promise<number>;
29	};
30
31	type IDataPrismaScopedClient = IDataPrismaExecutor & {
32	$tx?: <T>(fn: (prisma: IDataPrismaExecutor) => Promise<T>) => Promise<T>;
33	$transaction?: <T>(fn: (prisma: IDataPrismaExecutor) => Promise<T>) => Promise<T>;
34	};
35
36	interface IBaseImportJunctionCsvJob {
37	path: string;
38	baseId: string;
39	tableIdMap: Record<string, string>;
40	fieldIdMap: Record<string, string>;
41	structure: IBaseJson;
42	}
43
44	export const BASE_IMPORT_JUNCTION_CSV_QUEUE = 'base-import-junction-csv-queue';	181✔
45
46	@Injectable()
47	@Processor(BASE_IMPORT_JUNCTION_CSV_QUEUE)
48	export class BaseImportJunctionCsvQueueProcessor extends WorkerHost {
49	private logger = new Logger(BaseImportJunctionCsvQueueProcessor.name);	194✔
50	private processedJobs = new Set<string>();	194✔
51
52	constructor(
53	private readonly prismaService: PrismaService,
54	private readonly persistedComputedBackfillService: PersistedComputedBackfillService,
55	@InjectStorageAdapter() private readonly storageAdapter: StorageAdapter,
56	@InjectQueue(BASE_IMPORT_JUNCTION_CSV_QUEUE)
57	public readonly queue: Queue<IBaseImportJunctionCsvJob>,
58	@InjectDbProvider() private readonly dbProvider: IDbProvider,
59	private readonly dataDbClientManager: DataDbClientManager
60	) {
61	super();	194✔
62	}
63
64	public async process(job: Job<IBaseImportJunctionCsvJob>) {
65	const jobId = String(job.id);	3✔
66	if (this.processedJobs.has(jobId)) {	3✔
67	this.logger.log(`Job ${jobId} already processed, skipping`);	×
68	return;	×
69	}
70
71	this.processedJobs.add(jobId);	3✔
72
73	const { path, baseId, tableIdMap, fieldIdMap, structure } = job.data;	3✔
74
75	try {	3✔
76	await this.importJunctionChunk(path, baseId, fieldIdMap, structure);	3✔
77	await this.persistedComputedBackfillService.recomputeForTables(Object.values(tableIdMap));	3✔
78	} catch (error) {
UNCOV 79	this.logger.error(	×
80	`Process base import junction csv failed: ${(error as Error)?.message}`,
81	(error as Error)?.stack
82	);
83	}
84	}
85
86	private async importJunctionChunk(
87	path: string,
88	baseId: string,
89	fieldIdMap: Record<string, string>,
90	structure: IBaseJson
91	) {
92	const csvStream = await this.storageAdapter.downloadFile(	3✔
93	StorageAdapter.getBucket(UploadType.Import),
94	path
95	);
96
97	const sourceLinkFields = structure.tables	3✔
98	.map(({ fields }) => fields)	6✔
99	.flat()
100	.filter((f) => f.type === FieldType.Link && !f.isLookup);	47✔
101
102	const linkFieldRaws = await this.prismaService.field.findMany({	3✔
103	where: {
104	id: {
105	in: Object.values(fieldIdMap),
106	},
107	type: FieldType.Link,
108	isLookup: null,
109	},
110	});
111
112	const junctionDbTableNameMap = {} as Record<	3✔
113	string,
114	{
115	sourceSelfKeyName: string;
116	sourceForeignKeyName: string;
117	targetSelfKeyName: string;
118	targetForeignKeyName: string;
119	targetFkHostTableName: string;
120	}
121	>;
122
123	const linkFieldInstances = linkFieldRaws.map((f) => createFieldInstanceByRaw(f));	10✔
124
125	for (const sourceField of sourceLinkFields) {	3✔
126	const { options: sourceOptions } = sourceField;	10✔
127	const {
128	fkHostTableName: sourceFkHostTableName,
129	selfKeyName: sourceSelfKeyName,
130	foreignKeyName: sourceForeignKeyName,
131	} = sourceOptions as ILinkFieldOptions;	10✔
132	const targetField = linkFieldInstances.find((f) => f.id === fieldIdMap[sourceField.id])!;	27✔
133	const { options: targetOptions } = targetField;	10✔
134	const {
135	fkHostTableName: targetFkHostTableName,
136	selfKeyName: targetSelfKeyName,
137	foreignKeyName: targetForeignKeyName,
138	} = targetOptions as ILinkFieldOptions;	10✔
139	if (sourceFkHostTableName.includes('junction_')) {	10✔
140	junctionDbTableNameMap[sourceFkHostTableName] = {	8✔
141	sourceSelfKeyName,
142	sourceForeignKeyName,
143	targetSelfKeyName,
144	targetForeignKeyName,
145	targetFkHostTableName,
146	};
147	}
148	}
149
150	const parser = unzipper.Parse();	3✔
151	csvStream.pipe(parser);	3✔
152
153	const processedFiles = new Set<string>();	3✔
154
155	return new Promise<{ success: boolean }>((resolve, reject) => {	3✔
156	parser.on('entry', (entry) => {	3✔
157	const filePath = entry.path;	5✔
158
159	if (processedFiles.has(filePath)) {	5✔
160	entry.autodrain();	×
161	return;	×
162	}
163	processedFiles.add(filePath);	5✔
164
165	if (	5✔
166	filePath.startsWith('tables/') &&
167	entry.type !== 'Directory' &&
168	filePath.includes('junction_')
169	) {
170	const name = filePath.replace('tables/', '').split('.');	×
171	name.pop();	×
172	const junctionTableName = name.join('.');	×
173	const junctionInfo = junctionDbTableNameMap[junctionTableName];	×
174
175	const {
176	sourceForeignKeyName,
177	targetForeignKeyName,
178	sourceSelfKeyName,
179	targetSelfKeyName,
180	targetFkHostTableName,
181	} = junctionInfo;	×
182
183	const batchProcessor = new BatchProcessor<Record<string, unknown>>((chunk) =>	×
184	this.handleJunctionChunk(baseId, chunk, targetFkHostTableName)	×
185	);
186
187	entry	×
188	.pipe(
189	csvParser.default({
190	// strict: true,
191	mapValues: ({ value }) => {
192	// deal with old junction order case
193	return value === '' ? null : value;	×
194	},
195	mapHeaders: ({ header }) => {
196	return header	×
197	.replaceAll(sourceForeignKeyName, targetForeignKeyName)
198	.replaceAll(sourceSelfKeyName, targetSelfKeyName);
199	},
200	})
201	)
202	.pipe(batchProcessor)
203	.on('error', (error: Error) => {
204	this.logger.error(`process csv import error: ${error.message}`, error.stack);	×
205	reject(error);	×
206	})
207	.on('end', () => {
208	this.logger.log(`csv ${junctionTableName} finished`);	×
209	});
210	} else {
211	entry.autodrain();	5✔
212	}
213	});
214
215	parser.on('close', () => {	3✔
216	this.logger.log('import csv junction completed');	3✔
217	resolve({ success: true });	3✔
218	});
219
220	parser.on('error', (error) => {	3✔
221	this.logger.error(`import csv junction parser error: ${error.message}`, error.stack);	×
222	reject(error);	×
223	});
224	});
225	}
226
227	private async dataTransaction<T>(
228	dataPrisma: IDataPrismaScopedClient,
229	fn: (prisma: IDataPrismaExecutor) => Promise<T>
230	) {
231	if (dataPrisma.$tx) {	×
232	return await dataPrisma.$tx(fn);	×
233	}
234
235	if (dataPrisma.$transaction) {	×
236	return await dataPrisma.$transaction(fn);	×
237	}
238
239	return await fn(dataPrisma);	×
240	}
241
242	private async handleJunctionChunk(
243	baseId: string,
244	results: Record<string, unknown>[],
245	targetFkHostTableName: string
246	) {
247	const allForeignKeyInfos = [] as {	×
248	constraint_name: string;
249	column_name: string;
250	referenced_table_schema: string;
251	referenced_table_name: string;
252	referenced_column_name: string;
253	dbTableName: string;
254	}[];
255
256	const dataPrisma = (await this.dataDbClientManager.dataPrismaForBase(	×
257	baseId
258	)) as IDataPrismaScopedClient;
259	const dataKnex = await this.dataDbClientManager.dataKnexForBase(baseId);	×
260
261	await this.dataTransaction(dataPrisma, async (prisma) => {	×
262	// delete foreign keys if(exist) then duplicate table data
263	const foreignKeysInfoSql = this.dbProvider.getForeignKeysInfo(targetFkHostTableName);	×
264	const foreignKeysInfo = await prisma.$queryRawUnsafe<	×
265	{
266	constraint_name: string;
267	column_name: string;
268	referenced_table_schema: string;
269	referenced_table_name: string;
270	referenced_column_name: string;
271	}[]
272	>(foreignKeysInfoSql);
273	const newForeignKeyInfos = foreignKeysInfo.map((info) => ({	×
274	...info,
275	dbTableName: targetFkHostTableName,
276	}));
277	allForeignKeyInfos.push(...newForeignKeyInfos);	×
278
279	for (const { constraint_name, column_name, dbTableName } of allForeignKeyInfos) {	×
280	const dropForeignKeyQuery = dataKnex.schema	×
281	.alterTable(dbTableName, (table) => {
282	table.dropForeign(column_name, constraint_name);	×
283	})
284	.toQuery();
285
286	await prisma.$executeRawUnsafe(dropForeignKeyQuery);	×
287	}
288
289	const sql = dataKnex.table(targetFkHostTableName).insert(results).toQuery();	×
290	try {	×
291	await prisma.$executeRawUnsafe(sql);	×
292	} catch (error) {
293	if (error instanceof PrismaClientKnownRequestError) {	×
294	this.logger.error(	×
295	`exc junction import task known error: (${error.code}): ${error.message}`,
296	error.stack
297	);
298	} else if (error instanceof PrismaClientUnknownRequestError) {	×
299	this.logger.error(	×
300	`exc junction import task unknown error: ${error.message}`,
301	error.stack
302	);
303	} else {
304	this.logger.error(	×
305	`exc junction import task error: ${(error as Error)?.message}`,
306	(error as Error)?.stack
307	);
308	}
309	}
310
311	// add foreign keys with NOT VALID to skip existing data validation
312	for (const {	×
313	constraint_name,
314	column_name,
315	dbTableName,
316	referenced_table_schema: referencedTableSchema,
317	referenced_table_name: referencedTableName,
318	referenced_column_name: referencedColumnName,
319	} of allForeignKeyInfos) {
320	const [schema, tableName] = dbTableName.split('.');	×
321	const addForeignKeyQuery = dataKnex	×
322	.raw(
323	'ALTER TABLE ??.?? ADD CONSTRAINT ?? FOREIGN KEY (??) REFERENCES ??.??(??) NOT VALID',
324	[
325	schema,
326	tableName,
327	constraint_name,
328	column_name,
329	referencedTableSchema,
330	referencedTableName,
331	referencedColumnName,
332	]
333	)
334	.toQuery();
335	await prisma.$executeRawUnsafe(addForeignKeyQuery);	×
336	}
337	});
338	}
339	}

teableio / teable / 29496424463

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous