-
-
Notifications
You must be signed in to change notification settings - Fork 1.4k
/
Copy pathMetaService.ts
137 lines (114 loc) · 3.41 KB
/
MetaService.ts
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
/*
* SPDX-FileCopyrightText: syuilo and other misskey contributors
* SPDX-License-Identifier: AGPL-3.0-only
*/
import { Inject, Injectable } from '@nestjs/common';
import { DataSource } from 'typeorm';
import * as Redis from 'ioredis';
import { DI } from '@/di-symbols.js';
import { MiMeta } from '@/models/entities/Meta.js';
import { GlobalEventService } from '@/core/GlobalEventService.js';
import { bindThis } from '@/decorators.js';
import { StreamMessages } from '@/server/api/stream/types.js';
import type { OnApplicationShutdown } from '@nestjs/common';
@Injectable()
export class MetaService implements OnApplicationShutdown {
private cache: MiMeta | undefined;
private intervalId: NodeJS.Timer;
constructor(
@Inject(DI.redisForSub)
private redisForSub: Redis.Redis,
@Inject(DI.db)
private db: DataSource,
private globalEventService: GlobalEventService,
) {
//this.onMessage = this.onMessage.bind(this);
if (process.env.NODE_ENV !== 'test') {
this.intervalId = setInterval(() => {
this.fetch(true).then(meta => {
// fetch内でもセットしてるけど仕様変更の可能性もあるため一応
this.cache = meta;
});
}, 1000 * 60 * 5);
}
this.redisForSub.on('message', this.onMessage);
}
@bindThis
private async onMessage(_: string, data: string): Promise<void> {
const obj = JSON.parse(data);
if (obj.channel === 'internal') {
const { type, body } = obj.message as StreamMessages['internal']['payload'];
switch (type) {
case 'metaUpdated': {
this.cache = body;
break;
}
default:
break;
}
}
}
@bindThis
public async fetch(noCache = false): Promise<MiMeta> {
if (!noCache && this.cache) return this.cache;
return await this.db.transaction(async transactionalEntityManager => {
// 過去のバグでレコードが複数出来てしまっている可能性があるので新しいIDを優先する
const metas = await transactionalEntityManager.find(MiMeta, {
order: {
id: 'DESC',
},
});
const meta = metas[0];
if (meta) {
this.cache = meta;
return meta;
} else {
// metaが空のときfetchMetaが同時に呼ばれるとここが同時に呼ばれてしまうことがあるのでフェイルセーフなupsertを使う
const saved = await transactionalEntityManager
.upsert(
MiMeta,
{
id: 'x',
},
['id'],
)
.then((x) => transactionalEntityManager.findOneByOrFail(MiMeta, x.identifiers[0]));
this.cache = saved;
return saved;
}
});
}
@bindThis
public async update(data: Partial<MiMeta>): Promise<MiMeta> {
const updated = await this.db.transaction(async transactionalEntityManager => {
const metas = await transactionalEntityManager.find(MiMeta, {
order: {
id: 'DESC',
},
});
const meta = metas[0];
if (meta) {
await transactionalEntityManager.update(MiMeta, meta.id, data);
const metas = await transactionalEntityManager.find(MiMeta, {
order: {
id: 'DESC',
},
});
return metas[0];
} else {
return await transactionalEntityManager.save(MiMeta, data);
}
});
this.globalEventService.publishInternalEvent('metaUpdated', updated);
return updated;
}
@bindThis
public dispose(): void {
clearInterval(this.intervalId);
this.redisForSub.off('message', this.onMessage);
}
@bindThis
public onApplicationShutdown(signal?: string | undefined): void {
this.dispose();
}
}