10317771079

Committed 09 Aug 2024 10:30AM UTC coverage: 82.67% (+0.001%) from 82.669%

Build # 10317771079

Build Type

Pull #810

github

Committed by

web-flow

Commit Message

Merge 1ca001cff into 76ca756bb

Pull Request Pull Request #810: fix: import relative

Run Details

4404 of 4626 branches covered (95.2%)

90 of 102 new or added lines in 6 files covered. (88.24%)

28 existing lines in 4 files now uncovered.

29329 of 35477 relevant lines covered (82.67%)

1241.38 hits per line

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

76.87

/apps/nestjs-backend/src/share-db/share-db.service.ts

import { Injectable, Logger } from '@nestjs/common';
import { context as otelContext, trace as otelTrace } from '@opentelemetry/api';
import { FieldOpBuilder, IdPrefix, ViewOpBuilder } from '@teable/core';
import { PrismaService } from '@teable/db-main-prisma';
import { noop } from 'lodash';
import { ClsService } from 'nestjs-cls';
import type { CreateOp, DeleteOp, EditOp } from 'sharedb';
import ShareDBClass from 'sharedb';
import { CacheConfig, ICacheConfig } from '../configs/cache.config';
import { EventEmitterService } from '../event-emitter/event-emitter.service';
import type { IClsStore } from '../types/cls';
import { Timing } from '../utils/timing';
import { authMiddleware } from './auth.middleware';
import type { IRawOpMap } from './interface';
import { ShareDbAdapter } from './share-db.adapter';
import { RedisPubSub } from './sharedb-redis.pubsub';

@Injectable()
export class ShareDbService extends ShareDBClass {
  private logger = new Logger(ShareDbService.name);

  constructor(
    readonly shareDbAdapter: ShareDbAdapter,
    private readonly eventEmitterService: EventEmitterService,
    private readonly prismaService: PrismaService,
    private readonly cls: ClsService<IClsStore>,
    @CacheConfig() private readonly cacheConfig: ICacheConfig
  ) {
    super({
      presence: true,
      doNotForwardSendPresenceErrorsToClient: true,
      db: shareDbAdapter,
    });

    const { provider, redis } = this.cacheConfig;
    if (provider === 'redis') {
      if (!redis.uri) {
        throw new Error('Redis URI is required for Redis cache provider.');
      }
      const redisPubsub = new RedisPubSub({ redisURI: redis.uri });

      this.logger.log(`> Detected Redis cache; enabled the Redis pub/sub adapter for ShareDB.`);
      this.pubsub = redisPubsub;
    }

    authMiddleware(this);
    this.use('submit', this.onSubmit);

    // broadcast raw op events to client
    this.prismaService.bindAfterTransaction(() => {
      const rawOpMaps = this.cls.get('tx.rawOpMaps');
      this.cls.set('tx.rawOpMaps', undefined);

      const ops: IRawOpMap[] = [];
      if (rawOpMaps?.length) {
        ops.push(...rawOpMaps);
      }

      if (ops.length) {
        this.publishOpsMap(rawOpMaps);
        this.eventEmitterService.ops2Event(ops);
      }
    });
  }

  getConnection() {
    return this.connect();
  }

  @Timing()
  publishOpsMap(rawOpMaps: IRawOpMap[] | undefined) {
    if (!rawOpMaps?.length) {
      return;
    }

    for (const rawOpMap of rawOpMaps) {
      for (const collection in rawOpMap) {
        const data = rawOpMap[collection];
        for (const docId in data) {
          const rawOp = data[docId] as EditOp | CreateOp | DeleteOp;
          const channels = [collection, `${collection}.${docId}`];
          rawOp.c = collection;
          rawOp.d = docId;
          this.pubsub.publish(channels, rawOp, noop);

          if (this.shouldPublishAction(rawOp)) {
            const tableId = collection.split('_')[1];
            this.publishRelatedChannels(tableId, rawOp);
          }
        }
      }
    }
  }

  private shouldPublishAction(rawOp: EditOp | CreateOp | DeleteOp) {
    const viewKeys = ['filter', 'sort', 'group', 'lastModifiedTime'];
    const fieldKeys = ['options'];
    return rawOp.op?.some(
      (op) =>
        viewKeys.includes(ViewOpBuilder.editor.setViewProperty.detect(op)?.key as string) ||
        fieldKeys.includes(FieldOpBuilder.editor.setFieldProperty.detect(op)?.key as string)
    );
  }

  /**
   * this is for some special scenarios like manual sort
   * which only send view ops but update record too
   */
  private publishRelatedChannels(tableId: string, rawOp: EditOp | CreateOp | DeleteOp) {
    this.pubsub.publish([`${IdPrefix.Record}_${tableId}`], rawOp, noop);
    this.pubsub.publish([`${IdPrefix.Field}_${tableId}`], rawOp, noop);
  }

  private onSubmit = (
    context: ShareDBClass.middleware.SubmitContext,
    next: (err?: unknown) => void
  ) => {
    const tracer = otelTrace.getTracer('default');
    const currentSpan = tracer.startSpan('submitOp');

    // console.log('onSubmit start');

    otelContext.with(otelTrace.setSpan(otelContext.active(), currentSpan), () => {
      const [docType] = context.collection.split('_');

      if (docType !== IdPrefix.Record || !context.op.op) {
        return next(new Error('only record op can be committed'));
      }
      next();
    });

    // console.log('onSubmit end');
  };
}

1	import { Injectable, Logger } from '@nestjs/common';	4✔
2	import { context as otelContext, trace as otelTrace } from '@opentelemetry/api';	4✔
3	import { FieldOpBuilder, IdPrefix, ViewOpBuilder } from '@teable/core';	4✔
4	import { PrismaService } from '@teable/db-main-prisma';	4✔
5	import { noop } from 'lodash';	4✔
6	import { ClsService } from 'nestjs-cls';	4✔
7	import type { CreateOp, DeleteOp, EditOp } from 'sharedb';	4✔
8	import ShareDBClass from 'sharedb';	4✔
9	import { CacheConfig, ICacheConfig } from '../configs/cache.config';	4✔
10	import { EventEmitterService } from '../event-emitter/event-emitter.service';	4✔
11	import type { IClsStore } from '../types/cls';	4✔
12	import { Timing } from '../utils/timing';	4✔
13	import { authMiddleware } from './auth.middleware';	4✔
14	import type { IRawOpMap } from './interface';	4✔
15	import { ShareDbAdapter } from './share-db.adapter';	4✔
16	import { RedisPubSub } from './sharedb-redis.pubsub';	4✔
17		4✔
18	@Injectable()	4✔
19	export class ShareDbService extends ShareDBClass {	4✔
20	private logger = new Logger(ShareDbService.name);	364✔
21		364✔
22	constructor(	364✔
23	readonly shareDbAdapter: ShareDbAdapter,	364✔
24	private readonly eventEmitterService: EventEmitterService,	364✔
25	private readonly prismaService: PrismaService,	364✔
26	private readonly cls: ClsService<IClsStore>,	364✔
27	@CacheConfig() private readonly cacheConfig: ICacheConfig	364✔
28	) {	364✔
29	super({	364✔
30	presence: true,	364✔
31	doNotForwardSendPresenceErrorsToClient: true,	364✔
32	db: shareDbAdapter,	364✔
33	});	364✔
34		364✔
35	const { provider, redis } = this.cacheConfig;	364✔
36	if (provider === 'redis') {	364!
37	if (!redis.uri) {	×
38	throw new Error('Redis URI is required for Redis cache provider.');	×
39	}	×
40	const redisPubsub = new RedisPubSub({ redisURI: redis.uri });	×
41		×
42	this.logger.log(`> Detected Redis cache; enabled the Redis pub/sub adapter for ShareDB.`);	×
43	this.pubsub = redisPubsub;	×
44	}	×
45		364✔
46	authMiddleware(this);	364✔
47	this.use('submit', this.onSubmit);	364✔
48		364✔
49	// broadcast raw op events to client	364✔
50	this.prismaService.bindAfterTransaction(() => {	364✔
51	const rawOpMaps = this.cls.get('tx.rawOpMaps');	9,200✔
52	this.cls.set('tx.rawOpMaps', undefined);	9,200✔
53		9,200✔
54	const ops: IRawOpMap[] = [];	9,200✔
55	if (rawOpMaps?.length) {	9,200✔
56	ops.push(...rawOpMaps);	5,727✔
57	}	5,727✔
58		9,200✔
59	if (ops.length) {	9,200✔
60	this.publishOpsMap(rawOpMaps);	5,727✔
61	this.eventEmitterService.ops2Event(ops);	5,727✔
62	}	5,727✔
63	});	9,200✔
64	}	364✔
65		364✔
66	getConnection() {	364✔
67	return this.connect();	×
68	}	×
69		364✔
70	@Timing()	364✔
71	publishOpsMap(rawOpMaps: IRawOpMap[] \| undefined) {	364✔
72	if (!rawOpMaps?.length) {	5,727✔
73	return;	×
74	}	×
75		5,727✔
76	for (const rawOpMap of rawOpMaps) {	5,727✔
77	for (const collection in rawOpMap) {	12,510✔
78	const data = rawOpMap[collection];	12,510✔
79	for (const docId in data) {	12,510✔
80	const rawOp = data[docId] as EditOp \| CreateOp \| DeleteOp;	62,259✔
81	const channels = [collection, `${collection}.${docId}`];	62,259✔
82	rawOp.c = collection;	62,259✔
83	rawOp.d = docId;	62,259✔
84	this.pubsub.publish(channels, rawOp, noop);	62,259✔
85		62,259✔
86	if (this.shouldPublishAction(rawOp)) {	62,259✔
87	const tableId = collection.split('_')[1];	246✔
88	this.publishRelatedChannels(tableId, rawOp);	246✔
89	}	246✔
90	}	62,259✔
91	}	12,510✔
92	}	12,510✔
93	}	5,727✔
94		364✔
95	private shouldPublishAction(rawOp: EditOp \| CreateOp \| DeleteOp) {	364✔
96	const viewKeys = ['filter', 'sort', 'group', 'lastModifiedTime'];	62,259✔
97	const fieldKeys = ['options'];	62,259✔
98	return rawOp.op?.some(	62,259✔
99	(op) =>	62,259✔
100	viewKeys.includes(ViewOpBuilder.editor.setViewProperty.detect(op)?.key as string) \|\|	53,798✔
101	fieldKeys.includes(FieldOpBuilder.editor.setFieldProperty.detect(op)?.key as string)	53,684✔
102	);	62,259✔
103	}	62,259✔
104		364✔
105	/**	364✔
106	* this is for some special scenarios like manual sort	364✔
107	* which only send view ops but update record too	364✔
108	*/	364✔
109	private publishRelatedChannels(tableId: string, rawOp: EditOp \| CreateOp \| DeleteOp) {	364✔
110	this.pubsub.publish([`${IdPrefix.Record}_${tableId}`], rawOp, noop);	246✔
111	this.pubsub.publish([`${IdPrefix.Field}_${tableId}`], rawOp, noop);	246✔
112	}	246✔
113		364✔
114	private onSubmit = (	364✔
UNCOV 115	context: ShareDBClass.middleware.SubmitContext,	×
UNCOV 116	next: (err?: unknown) => void	×
UNCOV 117	) => {	×
UNCOV 118	const tracer = otelTrace.getTracer('default');	×
UNCOV 119	const currentSpan = tracer.startSpan('submitOp');	×
120		×
121	// console.log('onSubmit start');	×
122		×
123	otelContext.with(otelTrace.setSpan(otelContext.active(), currentSpan), () => {	×
124	const [docType] = context.collection.split('_');	×
125		×
126	if (docType !== IdPrefix.Record \|\| !context.op.op) {	×
127	return next(new Error('only record op can be committed'));	×
128	}	×
129	next();	×
130	});	×
131		×
132	// console.log('onSubmit end');	×
133	};	×
134	}	364✔

teableio / teable / 10317771079

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