flush all KvStorage-s before quitting
Massimo Melina committed
May 30, 2024 at 12:11 UTC
5d2636ab44b2662bf2bf6f2029414cd2ebf7af3b
7 files changed
+46
-30
src/fileAttr.ts
+2
@@ -3,10 +3,12 @@ import { existsSync } from 'fs'
3
import { promisify } from 'util'
4
import { access } from 'fs/promises'
5
import { tryJson } from './cross'
6
+import { onProcessExit } from './first'
7
// @ts-ignore
8
import fsx from 'fs-x-attributes'
9
10
let fileAttrDb = new KvStorage({ defaultPutDelay: 1000, maxPutDelay: 5000 })
11
+onProcessExit(() => fileAttrDb.flush())
12
const FN = 'file-attr.kv'
13
if (existsSync(FN))
14
fileAttrDb.open(FN)
src/first.ts
new
+23
@@ -0,0 +1,23 @@
1
+// should not import other sources that themselves import this file, to avoid circular dependencies
2
+import { EventEmitter } from 'events'
3
+
4
+type ProcessExitHandler = (signal:string) => any
5
+const cbs = new Set<ProcessExitHandler>()
6
+export function onProcessExit(cb: ProcessExitHandler) {
7
+ cbs.add(cb)
8
+ return () => cbs.delete(cb)
9
+}
10
+onFirstEvent(process, ['exit', 'SIGQUIT', 'SIGTERM', 'SIGINT', 'SIGHUP'], signal =>
11
+ Promise.allSettled(Array.from(cbs).map(cb => cb(signal))).then(() =>
12
+ process.exit(0)))
13
+
14
+export function onFirstEvent(emitter:EventEmitter, events: string[], cb: (...args:any[])=> void) {
15
+ let already = false
16
+ for (const e of events)
17
+ emitter.once(e, (...args) => {
18
+ if (already) return
19
+ already = true
20
+ cb(...args)
21
+ })
22
+}
23
+
src/ip.ts
+2
@@ -4,6 +4,7 @@ import { defineConfig } from './config'
4
import { KvStorage } from '@rejetto/kvstorage'
5
import { Middleware } from 'koa'
6
import { CFG, isLocalHost, MINUTE } from './misc'
7
+import { onProcessExit } from './first'
8
9
const trackIps = defineConfig(CFG.track_ips, true)
10
export const ips = new KvStorage({
@@ -11,6 +12,7 @@ export const ips = new KvStorage({
12
maxPutDelay: 10 * MINUTE,
13
maxPutDelayCreate: 0,
14
})
15
+onProcessExit(() => ips.flush())
16
17
export const trackIpsMw: Middleware = async (ctx, next) => {
18
if (trackIps.get() && !isLocalHost(ctx))
src/misc.ts
-21
@@ -1,6 +1,5 @@
1
// This file is part of HFS - Copyright 2021-2023, Massimo Melina <a@rejetto.com> - License https://www.gnu.org/licenses/gpl-3.0.txt
2
3
-import { EventEmitter } from 'events'
3
import { basename } from 'path'
4
import Koa from 'koa'
5
import { Connection } from './connections'
@@ -17,26 +16,6 @@ import { HTTP_BAD_REQUEST } from './const'
16
import { isIpLocalHost, makeMatcher } from './cross'
17
import { isIPv6 } from 'net'
18
20
-type ProcessExitHandler = (signal:string) => any
21
-const cbs = new Set<ProcessExitHandler>()
22
-export function onProcessExit(cb: ProcessExitHandler) {
23
- cbs.add(cb)
24
- return () => cbs.delete(cb)
25
-}
26
-onFirstEvent(process, ['exit', 'SIGQUIT', 'SIGTERM', 'SIGINT', 'SIGHUP'], signal =>
27
- Promise.allSettled(Array.from(cbs).map(cb => cb(signal))).then(() =>
28
- process.exit(0)))
29
-
30
-export function onFirstEvent(emitter:EventEmitter, events: string[], cb: (...args:any[])=> void) {
31
- let already = false
32
- for (const e of events)
33
- emitter.once(e, (...args) => {
34
- if (already) return
35
- already = true
36
- cb(...args)
37
- })
38
-}
39
-
19
export function pattern2filter(pattern: string){
20
const matcher = makeMatcher(pattern.includes('*') ? pattern // if you specify *, we'll respect its position
21
: pattern.split('|').map(x => `*${x}*`).join('|'))
src/persistence.ts
+2
-1
@@ -1,5 +1,6 @@
1
import { KvStorage } from '@rejetto/kvstorage'
2
-import { MINUTE, onProcessExit } from './misc'
2
+import { MINUTE } from './misc'
3
+import { onProcessExit } from './first'
4
5
export const storedMap = new KvStorage({
6
defaultPutDelay: 5000,
src/plugins.ts
+15
-7
@@ -7,7 +7,7 @@ import { API_VERSION, APP_PATH, COMPATIBLE_API_VERSION, HTTP_NOT_FOUND, IS_WINDO
7
PLUGINS_PUB_URI } from './const'
8
import * as Const from './const'
9
import Koa from 'koa'
10
-import { adjustStaticPathForGlob, Callback, debounceAsync, Dict, getOrSet, onlyTruthy, onProcessExit,
10
+import { adjustStaticPathForGlob, Callback, debounceAsync, Dict, getOrSet, onlyTruthy,
11
PendingPromise, pendingPromise, Promisable, same, tryJson, wait, waitFor, wantArray, watchDir } from './misc'
12
import { defineConfig, getConfig } from './config'
13
import { DirEntry } from './api.get_file_list'
@@ -20,6 +20,7 @@ import { getConnections } from './connections'
20
import { dirname, join, resolve } from 'path'
21
import { watchLoadCustomHtml } from './customHtml'
22
import { KvStorage, KvStorageOptions } from '@rejetto/kvstorage'
23
+import { onProcessExit } from './first'
24
25
export const PATH = 'plugins'
26
export const DISABLING_SUFFIX = '-disabled'
@@ -98,7 +99,7 @@ export function getPluginConfigFields(id: string) {
99
return plugins[id]?.getData().config
100
}
101
101
-async function initPlugin<T>(pl: any, more?: T) {
102
+async function initPlugin<T>(pl: any, morePassedToInit?: T) {
103
return Object.assign(pl, await pl.init?.({
104
Const,
105
require,
@@ -107,7 +108,7 @@ async function initPlugin<T>(pl: any, more?: T) {
108
log: console.log,
109
getHfsConfig: getConfig,
110
customApiCall,
110
- ...more
111
+ ...morePassedToInit
112
}))
113
}
114
@@ -166,7 +167,7 @@ type OnDirEntry = (params:OnDirEntryParams) => void | false
167
export class Plugin implements CommonPluginInterface {
168
started: Date | null = new Date()
169
169
- constructor(readonly id:string, readonly folder:string, private readonly data:any, private unwatch:()=>void){
170
+ constructor(readonly id:string, readonly folder:string, private readonly data:any, private onUnload:()=>unknown){
171
if (!data) throw 'invalid data'
172
173
this.data = data = { ...data } // clone to make object modifiable. Objects coming from import are not.
@@ -212,6 +213,7 @@ export class Plugin implements CommonPluginInterface {
213
const { id } = this
214
try {
215
await this.data?.unload?.()
216
+ await this.onUnload()
217
if (!reloading && id !== SERVER_CODE_ID) // we already printed 'reloading'
218
console.log('unloaded plugin', id)
219
}
@@ -220,7 +222,6 @@ export class Plugin implements CommonPluginInterface {
222
}
223
if (this.data)
224
this.data.unload = undefined
223
- this.unwatch()
225
}
226
}
227
@@ -390,6 +391,7 @@ function watchPlugin(id: string, path: string) {
391
console.debug("starting plugin", id)
392
const storageDir = resolve(module, '..', STORAGE_FOLDER) + (IS_WINDOWS ? '\\' : '/')
393
await mkdir(storageDir, { recursive: true })
394
+ const dbs: KvStorage[] = []
395
await initPlugin(pluginData, {
396
srcDir: __dirname,
397
storageDir,
@@ -397,6 +399,7 @@ function watchPlugin(id: string, path: string) {
399
if (!filename) throw Error("missing filename")
400
const db = new KvStorage(options)
401
await db.open(join(storageDir, filename))
402
+ dbs.push(db)
403
return db
404
},
405
log(...args: any[]) {
@@ -420,7 +423,12 @@ function watchPlugin(id: string, path: string) {
423
const folder = dirname(module)
424
const { state, unwatch } = watchLoadCustomHtml(folder)
425
pluginData.customHtml = state
423
- const plugin = new Plugin(id, folder, pluginData, unwatch)
426
+
427
+ const plugin = new Plugin(id, folder, pluginData, async () => {
428
+ unwatch()
429
+ await Promise.allSettled(dbs.map(x => x.flush()))
430
+ dbs.length = 0
431
+ })
432
if (alreadyRunning)
433
events.emit('pluginUpdated', Object.assign(_.pick(plugin, 'started'), getPluginInfo(id)))
434
else {
@@ -492,7 +500,7 @@ export function parsePluginSource(id: string, source: string) {
500
pl.apiRequired = tryJson(/exports.apiRequired *= *([ \d.,[\]]+)/.exec(source)?.[1]) ?? undefined
501
pl.isTheme = tryJson(/exports.isTheme *= *(true|false|"light"|"dark")/.exec(source)?.[1]) ?? (id.endsWith('-theme') || undefined)
502
pl.preview = tryJson(/exports.preview *= *(.+)/.exec(source)?.[1]) ?? undefined
495
- pl.depend = tryJson(/exports.depend *= *(\[.*\])/m.exec(source)?.[1])?.filter((x: any) =>
503
+ pl.depend = tryJson(/exports.depend *= *(\[.*])/m.exec(source)?.[1])?.filter((x: any) =>
504
typeof x.repo === 'string' && x.version === undefined || typeof x.version === 'number'
505
|| console.warn("plugin dependency discarded", x) )
506
if (Array.isArray(pl.apiRequired) && (pl.apiRequired.length !== 2 || !pl.apiRequired.every(_.isFinite))) // validate [from,to] form
src/update.ts
+2
-1
@@ -4,7 +4,7 @@ import { getRepoInfo } from './github'
4
import { argv, HFS_REPO, IS_BINARY, IS_WINDOWS, RUNNING_BETA } from './const'
5
import { dirname, join } from 'path'
6
import { spawn, spawnSync } from 'child_process'
7
-import { httpStream, onProcessExit, unzip } from './misc'
7
+import { httpStream, unzip } from './misc'
8
import { createReadStream, renameSync, unlinkSync } from 'fs'
9
import { pluginsWatcher } from './plugins'
10
import { access, chmod, stat } from 'fs/promises'
@@ -12,6 +12,7 @@ import { Readable } from 'stream'
12
import open from 'open'
13
import { currentVersion, defineConfig, versionToScalar } from './config'
14
import { RUNNING_AS_SERVICE } from './util-os'
15
+import { onProcessExit } from './first'
16
17
const updateToBeta = defineConfig('update_to_beta', false)
18