mirror of https://github.com/nocodb/nocodb
mertmit
1 year ago
18 changed files with 679 additions and 427 deletions
@ -1,102 +0,0 @@
|
||||
import type { Socket } from 'socket.io-client' |
||||
import io from 'socket.io-client' |
||||
import { JobStatus, defineNuxtPlugin, useGlobal, watch } from '#imports' |
||||
|
||||
export default defineNuxtPlugin(async (nuxtApp) => { |
||||
const { appInfo } = useGlobal() |
||||
|
||||
let socket: Socket | null = null |
||||
let messageIndex = 0 |
||||
|
||||
const init = async (token: string) => { |
||||
try { |
||||
if (socket) socket.disconnect() |
||||
|
||||
const url = new URL(appInfo.value.ncSiteUrl, window.location.href.split(/[?#]/)[0]) |
||||
let socketPath = url.pathname |
||||
socketPath += socketPath.endsWith('/') ? 'socket.io' : '/socket.io' |
||||
|
||||
socket = io(`${url.href}jobs`, { |
||||
extraHeaders: { 'xc-auth': token }, |
||||
path: socketPath, |
||||
}) |
||||
|
||||
socket.on('connect_error', (e) => { |
||||
console.error(e) |
||||
socket?.disconnect() |
||||
}) |
||||
} catch {} |
||||
} |
||||
|
||||
if (nuxtApp.$state.signedIn.value) { |
||||
await init(nuxtApp.$state.token.value) |
||||
} |
||||
|
||||
const send = (evt: string, data: any) => { |
||||
if (socket) { |
||||
const _id = messageIndex++ |
||||
socket.emit(evt, { _id, data }) |
||||
return _id |
||||
} |
||||
} |
||||
|
||||
const jobs = { |
||||
subscribe( |
||||
job: { id: string } | any, |
||||
subscribedCb?: () => void, |
||||
statusCb?: (status: JobStatus, data?: any) => void, |
||||
logCb?: (data: { message: string }) => void, |
||||
) { |
||||
const logFn = (data: { id: string; data: { message: string } }) => { |
||||
if (data.id === job.id) { |
||||
if (logCb) logCb(data.data) |
||||
} |
||||
} |
||||
const statusFn = (data: any) => { |
||||
if (data.id === job.id) { |
||||
if (statusCb) statusCb(data.status, data.data) |
||||
if (data.status === JobStatus.COMPLETED || data.status === JobStatus.FAILED) { |
||||
socket?.off('status', statusFn) |
||||
socket?.off('log', logFn) |
||||
} |
||||
} |
||||
} |
||||
|
||||
const _id = send('subscribe', job) |
||||
|
||||
const subscribeFn = (data: { _id: number; id: string }) => { |
||||
if (data._id === _id) { |
||||
if (data.id !== job.id) { |
||||
job.id = data.id |
||||
} |
||||
if (subscribedCb) subscribedCb() |
||||
socket?.on('log', logFn) |
||||
socket?.on('status', statusFn) |
||||
socket?.off('subscribed', subscribeFn) |
||||
} |
||||
} |
||||
socket?.on('subscribed', subscribeFn) |
||||
}, |
||||
getStatus(id: string): Promise<string> { |
||||
return new Promise((resolve) => { |
||||
if (socket) { |
||||
const _id = send('status', { id }) |
||||
const tempFn = (data: any) => { |
||||
if (data._id === _id) { |
||||
resolve(data.status) |
||||
socket?.off('status', tempFn) |
||||
} |
||||
} |
||||
socket.on('status', tempFn) |
||||
} |
||||
}) |
||||
}, |
||||
} |
||||
|
||||
watch((nuxtApp.$state as ReturnType<typeof useGlobal>).token, (newToken, oldToken) => { |
||||
if (newToken && newToken !== oldToken) init(newToken) |
||||
else if (!newToken) socket?.disconnect() |
||||
}) |
||||
|
||||
nuxtApp.provide('jobs', jobs) |
||||
}) |
||||
@ -0,0 +1,92 @@
|
||||
import type { Api as BaseAPI } from 'nocodb-sdk' |
||||
import { defineNuxtPlugin } from '#imports' |
||||
|
||||
export default defineNuxtPlugin(async (nuxtApp) => { |
||||
const api: BaseAPI<any> = nuxtApp.$api as any |
||||
|
||||
// unsubscribe all if signed out
|
||||
let unsub = false |
||||
|
||||
const subscribe = async ( |
||||
topic: { id: string } | any, |
||||
cb: (data: { |
||||
id: string |
||||
status?: string |
||||
data?: { |
||||
error?: { |
||||
message: string |
||||
} |
||||
message?: string |
||||
result?: any |
||||
} |
||||
}) => void, |
||||
_mid = 0, |
||||
) => { |
||||
if (unsub) return |
||||
|
||||
try { |
||||
const response: |
||||
| { |
||||
_mid: number |
||||
id: string |
||||
status: 'refresh' | 'update' | 'close' |
||||
data: any |
||||
} |
||||
| { |
||||
_mid: number |
||||
id: string |
||||
status: 'refresh' | 'update' | 'close' |
||||
data: any |
||||
}[] = await api.jobs.listen({ _mid, data: topic }) |
||||
|
||||
if (Array.isArray(response)) { |
||||
let lastMid = 0 |
||||
for (const r of response) { |
||||
if (r.status === 'close') { |
||||
return cb(r) |
||||
} else { |
||||
if (r.status === 'update') { |
||||
cb(r.data) |
||||
} |
||||
lastMid = r._mid |
||||
} |
||||
} |
||||
await subscribe(topic, cb, lastMid) |
||||
} else { |
||||
if (response.status === 'close') { |
||||
return cb(response) |
||||
} else if (response.status === 'update') { |
||||
cb(response.data) |
||||
await subscribe(topic, cb, response._mid) |
||||
} else if (response.status === 'refresh') { |
||||
await subscribe(topic, cb, _mid) |
||||
} |
||||
} |
||||
} catch (e) { |
||||
setTimeout(() => { |
||||
subscribe(topic, cb, _mid) |
||||
}, 1000) |
||||
} |
||||
} |
||||
|
||||
const init = () => { |
||||
unsub = false |
||||
} |
||||
|
||||
if ((nuxtApp.$state as ReturnType<typeof useGlobal>).signedIn.value) { |
||||
await init() |
||||
} |
||||
|
||||
watch((nuxtApp.$state as ReturnType<typeof useGlobal>).token, (newToken, oldToken) => { |
||||
if (newToken && newToken !== oldToken) init() |
||||
else if (!newToken) { |
||||
unsub = true |
||||
} |
||||
}) |
||||
|
||||
const poller = { |
||||
subscribe, |
||||
} |
||||
|
||||
nuxtApp.provide('poller', poller) |
||||
}) |
||||
@ -0,0 +1,266 @@
|
||||
import { |
||||
Body, |
||||
Controller, |
||||
HttpCode, |
||||
Inject, |
||||
Post, |
||||
Request, |
||||
Response, |
||||
UseGuards, |
||||
} from '@nestjs/common'; |
||||
import { OnEvent } from '@nestjs/event-emitter'; |
||||
import { customAlphabet } from 'nanoid'; |
||||
import { JobsRedisService } from './redis/jobs-redis.service'; |
||||
import { JobStatus } from '~/interface/Jobs'; |
||||
import { JobEvents } from '~/interface/Jobs'; |
||||
import { GlobalGuard } from '~/guards/global/global.guard'; |
||||
import NocoCache from '~/cache/NocoCache'; |
||||
import { CacheDelDirection, CacheGetType, CacheScope } from '~/utils/globals'; |
||||
|
||||
const nanoidv2 = customAlphabet('1234567890abcdefghijklmnopqrstuvwxyz', 14); |
||||
const POLLING_INTERVAL = 10000; |
||||
|
||||
@Controller() |
||||
@UseGuards(GlobalGuard) |
||||
export class JobsController { |
||||
constructor( |
||||
@Inject('JobsService') private readonly jobsService, |
||||
private readonly jobsRedisService: JobsRedisService, |
||||
) {} |
||||
|
||||
private jobRooms = {}; |
||||
private localJobs = {}; |
||||
private closedJobs = []; |
||||
|
||||
@Post('/jobs/listen') |
||||
@HttpCode(200) |
||||
async listen( |
||||
@Response() res, |
||||
@Request() req, |
||||
@Body() body: { _mid: number; data: { id: string } }, |
||||
) { |
||||
const { _mid = 0, data } = body; |
||||
|
||||
const jobId = data.id; |
||||
|
||||
res.setHeader('Cache-Control', 'no-cache, must-revalidate'); |
||||
res.resId = nanoidv2(); |
||||
|
||||
let messages; |
||||
|
||||
if (this.localJobs[jobId]) { |
||||
messages = this.localJobs[jobId].messages; |
||||
} else { |
||||
messages = ( |
||||
await NocoCache.get( |
||||
`${CacheScope.JOBS}:${jobId}:messages`, |
||||
CacheGetType.TYPE_OBJECT, |
||||
) |
||||
)?.messages; |
||||
} |
||||
|
||||
const newMessages: any[] = []; |
||||
|
||||
if (messages) { |
||||
messages.forEach((m) => { |
||||
if (m._mid > _mid) { |
||||
newMessages.push(m); |
||||
} |
||||
}); |
||||
} |
||||
|
||||
if (newMessages.length > 0) { |
||||
res.send(newMessages); |
||||
return; |
||||
} |
||||
|
||||
if (this.closedJobs.includes(jobId)) { |
||||
res.send({ |
||||
status: 'close', |
||||
}); |
||||
return; |
||||
} |
||||
|
||||
if (this.jobRooms[jobId]) { |
||||
this.jobRooms[jobId].listeners.push(res); |
||||
} else { |
||||
this.jobRooms[jobId] = { |
||||
listeners: [res], |
||||
}; |
||||
// subscribe to job events
|
||||
this.jobsRedisService.subscribe(jobId, (data) => { |
||||
if (this.jobRooms[jobId]) { |
||||
this.jobRooms[jobId].listeners.forEach((res) => { |
||||
if (!res.headersSent) { |
||||
res.send({ |
||||
status: 'refresh', |
||||
}); |
||||
} |
||||
}); |
||||
} |
||||
|
||||
const cmd = data.cmd; |
||||
delete data.cmd; |
||||
switch (cmd) { |
||||
case JobEvents.STATUS: |
||||
if ([JobStatus.COMPLETED, JobStatus.FAILED].includes(data.status)) { |
||||
this.jobsRedisService.unsubscribe(jobId); |
||||
delete this.jobRooms[jobId]; |
||||
this.closedJobs.push(jobId); |
||||
setTimeout(() => { |
||||
this.closedJobs = this.closedJobs.filter((j) => j !== jobId); |
||||
}, POLLING_INTERVAL * 2); |
||||
} |
||||
break; |
||||
} |
||||
}); |
||||
} |
||||
|
||||
res.on('close', () => { |
||||
if (jobId && this.jobRooms[jobId]?.listeners) { |
||||
this.jobRooms[jobId].listeners = this.jobRooms[jobId].listeners.filter( |
||||
(r) => r.resId !== res.resId, |
||||
); |
||||
} |
||||
}); |
||||
|
||||
setTimeout(() => { |
||||
if (!res.headersSent) { |
||||
res.send({ |
||||
status: 'refresh', |
||||
}); |
||||
} |
||||
}, POLLING_INTERVAL); |
||||
} |
||||
|
||||
@Post('/jobs/status') |
||||
async status(@Body() data: { id: string } | any) { |
||||
let res: { |
||||
id?: string; |
||||
status?: JobStatus; |
||||
} | null = null; |
||||
if (Object.keys(data).every((k) => ['id'].includes(k)) && data?.id) { |
||||
const rooms = (await this.jobsService.jobList()).map( |
||||
(j) => `jobs-${j.id}`, |
||||
); |
||||
const room = rooms.find((r) => r === `jobs-${data.id}`); |
||||
if (room) { |
||||
res.id = data.id; |
||||
} |
||||
} else { |
||||
const job = await this.jobsService.getJobWithData(data); |
||||
if (job) { |
||||
res = {}; |
||||
res.id = job.id; |
||||
res.status = await this.jobsService.jobStatus(data.id); |
||||
} |
||||
} |
||||
|
||||
return res; |
||||
} |
||||
|
||||
@OnEvent(JobEvents.STATUS) |
||||
sendJobStatus(data: { id: string; status: JobStatus; data?: any }): void { |
||||
let response; |
||||
|
||||
const jobId = data.id; |
||||
|
||||
if (this.localJobs[jobId]) { |
||||
response = { |
||||
status: 'update', |
||||
data, |
||||
_mid: this.localJobs[jobId].messages.length + 1, |
||||
}; |
||||
this.localJobs[jobId].messages.push(response); |
||||
NocoCache.set(`${CacheScope.JOBS}:${jobId}:messages`, { |
||||
messages: this.localJobs[jobId].messages, |
||||
}); |
||||
} else { |
||||
response = { |
||||
status: 'update', |
||||
data, |
||||
_mid: 1, |
||||
}; |
||||
this.localJobs[jobId] = { |
||||
messages: [response], |
||||
}; |
||||
NocoCache.set(`${CacheScope.JOBS}:${jobId}:messages`, { |
||||
messages: this.localJobs[jobId].messages, |
||||
}); |
||||
} |
||||
|
||||
if (this.jobRooms[jobId]) { |
||||
this.jobRooms[jobId].listeners.forEach((res) => { |
||||
if (!res.headersSent) { |
||||
res.send(response); |
||||
} |
||||
}); |
||||
} |
||||
|
||||
if (process.env.NC_WORKER_CONTAINER === 'true') { |
||||
this.jobsRedisService.publish(jobId, { |
||||
cmd: JobEvents.STATUS, |
||||
...data, |
||||
}); |
||||
} |
||||
|
||||
if ([JobStatus.COMPLETED, JobStatus.FAILED].includes(data.status)) { |
||||
this.closedJobs.push(jobId); |
||||
setTimeout(() => { |
||||
this.closedJobs = this.closedJobs.filter((j) => j !== jobId); |
||||
}, POLLING_INTERVAL * 2); |
||||
|
||||
setTimeout(() => { |
||||
delete this.jobRooms[jobId]; |
||||
delete this.localJobs[jobId]; |
||||
NocoCache.deepDel(`jobs`, jobId, CacheDelDirection.CHILD_TO_PARENT); |
||||
}, POLLING_INTERVAL); |
||||
} |
||||
} |
||||
|
||||
@OnEvent(JobEvents.LOG) |
||||
sendJobLog(data: { id: string; data: { message: string } }): void { |
||||
let response; |
||||
|
||||
const jobId = data.id; |
||||
|
||||
if (this.localJobs[jobId]) { |
||||
response = { |
||||
status: 'update', |
||||
data, |
||||
_mid: this.localJobs[jobId].messages.length + 1, |
||||
}; |
||||
this.localJobs[jobId].messages.push(response); |
||||
NocoCache.set(`${CacheScope.JOBS}:${jobId}:messages`, { |
||||
messages: this.localJobs[jobId].messages, |
||||
}); |
||||
} else { |
||||
response = { |
||||
status: 'update', |
||||
data, |
||||
_mid: 1, |
||||
}; |
||||
this.localJobs[jobId] = { |
||||
messages: [response], |
||||
}; |
||||
NocoCache.set(`${CacheScope.JOBS}:${jobId}:messages`, { |
||||
messages: this.localJobs[jobId].messages, |
||||
}); |
||||
} |
||||
|
||||
if (this.jobRooms[jobId]) { |
||||
this.jobRooms[jobId].listeners.forEach((res) => { |
||||
if (!res.headersSent) { |
||||
res.send(response); |
||||
} |
||||
}); |
||||
} |
||||
|
||||
if (process.env.NC_WORKER_CONTAINER === 'true') { |
||||
this.jobsRedisService.publish(jobId, { |
||||
cmd: JobEvents.LOG, |
||||
...data, |
||||
}); |
||||
} |
||||
} |
||||
} |
||||
@ -1,110 +0,0 @@
|
||||
import { |
||||
ConnectedSocket, |
||||
MessageBody, |
||||
SubscribeMessage, |
||||
WebSocketGateway, |
||||
WebSocketServer, |
||||
} from '@nestjs/websockets'; |
||||
import { Server, Socket } from 'socket.io'; |
||||
import { ExecutionContextHost } from '@nestjs/core/helpers/execution-context-host'; |
||||
import { AuthGuard } from '@nestjs/passport'; |
||||
import { OnEvent } from '@nestjs/event-emitter'; |
||||
import { Inject } from '@nestjs/common'; |
||||
import type { OnModuleInit } from '@nestjs/common'; |
||||
import type { JobStatus } from '~/interface/Jobs'; |
||||
import { JobEvents } from '~/interface/Jobs'; |
||||
|
||||
const url = new URL( |
||||
process.env.NC_PUBLIC_URL || |
||||
`http://localhost:${process.env.PORT || '8080'}/`, |
||||
); |
||||
let namespace = url.pathname; |
||||
namespace += namespace.endsWith('/') ? 'jobs' : '/jobs'; |
||||
|
||||
@WebSocketGateway({ |
||||
cors: { |
||||
origin: '*', |
||||
allowedHeaders: ['xc-auth'], |
||||
credentials: true, |
||||
}, |
||||
namespace, |
||||
}) |
||||
export class JobsGateway implements OnModuleInit { |
||||
constructor(@Inject('JobsService') private readonly jobsService) {} |
||||
|
||||
@WebSocketServer() |
||||
server: Server; |
||||
|
||||
async onModuleInit() { |
||||
this.server.use(async (socket, next) => { |
||||
try { |
||||
const context = new ExecutionContextHost([socket.handshake as any]); |
||||
const guard = new (AuthGuard('jwt'))(context); |
||||
await guard.canActivate(context); |
||||
} catch {} |
||||
|
||||
next(); |
||||
}); |
||||
} |
||||
|
||||
@SubscribeMessage('subscribe') |
||||
async subscribe( |
||||
@MessageBody() |
||||
body: { _id: number; data: { id: string } | any }, |
||||
@ConnectedSocket() client: Socket, |
||||
): Promise<void> { |
||||
const { _id, data } = body; |
||||
if (Object.keys(data).every((k) => ['id'].includes(k)) && data?.id) { |
||||
const rooms = (await this.jobsService.jobList()).map( |
||||
(j) => `jobs-${j.id}`, |
||||
); |
||||
const room = rooms.find((r) => r === `jobs-${data.id}`); |
||||
if (room) { |
||||
client.join(`jobs-${data.id}`); |
||||
client.emit('subscribed', { |
||||
_id, |
||||
id: data.id, |
||||
}); |
||||
} |
||||
} else { |
||||
const job = await this.jobsService.getJobWithData(data); |
||||
if (job) { |
||||
client.join(`jobs-${job.id}`); |
||||
client.emit('subscribed', { |
||||
_id, |
||||
id: job.id, |
||||
}); |
||||
} |
||||
} |
||||
} |
||||
|
||||
@SubscribeMessage('status') |
||||
async status( |
||||
@MessageBody() body: { _id: number; data: { id: string } }, |
||||
@ConnectedSocket() client: Socket, |
||||
): Promise<void> { |
||||
const { _id, data } = body; |
||||
client.emit('status', { |
||||
_id, |
||||
id: data.id, |
||||
status: await this.jobsService.jobStatus(data.id), |
||||
}); |
||||
} |
||||
|
||||
@OnEvent(JobEvents.STATUS) |
||||
sendJobStatus(data: { id: string; status: JobStatus; data?: any }): void { |
||||
this.server.to(`jobs-${data.id}`).emit('status', { |
||||
id: data.id, |
||||
status: data.status, |
||||
data: data.data, |
||||
}); |
||||
} |
||||
|
||||
@OnEvent(JobEvents.LOG) |
||||
sendJobLog(data: { id: string; data: { message: string } }): void { |
||||
this.server.to(`jobs-${data.id}`).emit('log', { |
||||
id: data.id, |
||||
data: data.data, |
||||
}); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue