WEBDIENSTE/huly-platform
1
0
Fork 0
mirror of https://github.com/hcengineering/platform.git synced 2026-09-08 02:37:44 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge staging feb4 (#7910)

Browse Source
This commit is contained in:
Andrey Sobolev
2025-02-04 22:29:27 +07:00
committed by GitHub
parent 33cf7599cf
commit 04a282c0f8
21 changed files with 205 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.vscode/launch.json
Vendored
+5 -3
View File
@@ -295,7 +295,7 @@
"name": "Debug backup tool",
"type": "node",
"request": "launch",
"args": ["src/__start.ts", "backup-restore", "../../../hardware/dump/alex-staff-agency", "w-haiodo-alex-staff-c-673ee7ab-87df5406ea-2b8b4d", "--skip", "blob"],
"args": ["src/__start.ts", "backup", "../../../hardware/dump/githubcr", "w-haiodo-githubcr-67403799-de2a46aa46-beb3b2"],
"env": {
"MINIO_ACCESS_KEY": "minioadmin",
"MINIO_SECRET_KEY": "minioadmin",
@@ -305,12 +305,14 @@
"ACCOUNTS_URL": "http://localhost:3000",
"TELEGRAM_DATABASE": "telegram-service"
},
"smartStep": true,
"sourceMapRenames": true,
"runtimeArgs": ["--nolazy", "-r", "ts-node/register"],
"sourceMaps": true,
"cwd": "${workspaceRoot}/dev/tool",
"protocol": "inspector",
"outputCapture": "std",
"runtimeVersion": "20",
"runtimeVersion": "22",
"showAsyncStacks": true
},
{
@@ -401,7 +403,7 @@
"SECRET": "secret",
"REGION": "cockroach",
"BUCKET_NAME":"backups",
"INTERVAL":"43200"
"INTERVAL":"0"
},
"runtimeArgs": ["--nolazy", "-r", "ts-node/register"],
"showAsyncStacks": true,
common/scripts/version.txt
+1 -1
View File
@@ -1 +1 @@
"0.6.421"
"0.6.421"
packages/core/src/__tests__/client.test.ts
+5 -1
View File
@@ -24,7 +24,7 @@ import { ModelDb, TxDb } from '../memdb'
import { TxOperations } from '../operations'
import type { DocumentQuery, FindResult, SearchOptions, SearchQuery, SearchResult, TxResult } from '../storage'
import { Tx, TxFactory, TxProcessor } from '../tx'
import { fillConfiguration, pluginFilterTx } from '../utils'
import { fillConfiguration, generateId, pluginFilterTx } from '../utils'
import { connect } from './connection'
import { genMinModel } from './minmodel'
@@ -142,6 +142,10 @@ describe('client', () => {
finished: true
})
async getDomainHash (domain: Domain): Promise<string> {
return generateId()
}
async closeChunk (idx: number): Promise<void> {}
async loadDocs (domain: Domain, docs: Ref<Doc>[]): Promise<Doc[]> {
return []
packages/core/src/__tests__/connection.ts
+5 -1
View File
@@ -13,7 +13,7 @@
// limitations under the License.
//
import { ClientConnectEvent, DocChunk } from '..'
import { ClientConnectEvent, DocChunk, generateId } from '..'
import type { Class, Doc, Domain, Ref, Timestamp } from '../classes'
import { ClientConnection } from '../client'
import core from '../component'
@@ -82,6 +82,10 @@ export async function connect (handler: (tx: Tx) => void): Promise<ClientConnect
}
}
async getDomainHash (domain: Domain): Promise<string> {
return generateId()
}
async closeChunk (idx: number): Promise<void> {}
async loadDocs (domain: Domain, docs: Ref<Doc>[]): Promise<Doc[]> {
return []
packages/core/src/backup.ts
+2
View File
@@ -26,5 +26,7 @@ export interface BackupClient {
upload: (domain: Domain, docs: Doc[]) => Promise<void>
clean: (domain: Domain, docs: Ref<Doc>[]) => Promise<void>
getDomainHash: (domain: Domain) => Promise<string>
sendForceClose: () => Promise<void>
}
packages/core/src/client.ts
+4
View File
@@ -174,6 +174,10 @@ class ClientImpl implements Client, BackupClient {
return await this.conn.loadChunk(domain, idx)
}
async getDomainHash (domain: Domain): Promise<string> {
return await this.conn.getDomainHash(domain)
}
async closeChunk (idx: number): Promise<void> {
await this.conn.closeChunk(idx)
}
packages/core/src/server.ts
+2
View File
@@ -95,6 +95,8 @@ export interface LowLevelStorage {
query: DocumentQuery<T>,
options?: Pick<FindOptions<T>, 'sort' | 'limit' | 'projection'>
) => Promise<Iterator<T>>
getDomainHash: (ctx: MeasureContext, domain: Domain) => Promise<string>
}
export interface Iterator<T extends Doc> {
packages/query/src/__tests__/connection.ts
+5
View File
@@ -27,6 +27,7 @@ import core, {
FindOptions,
FindResult,
FulltextStorage,
generateId,
Hierarchy,
LoadModelResponse,
ModelDb,
@@ -121,6 +122,10 @@ FulltextStorage & {
}
}
async getDomainHash (domain: Domain): Promise<string> {
return generateId()
}
async loadModel (lastTxTime: Timestamp): Promise<Tx[]> {
return txes
}
plugins/client-resources/src/connection.ts
+4
View File
@@ -804,6 +804,10 @@ class Connection implements ClientConnection {
return this.sendRequest({ method: 'loadChunk', params: [domain, idx] })
}
async getDomainHash (domain: Domain): Promise<string> {
return await this.sendRequest({ method: 'getDomainHash', params: [domain] })
}
closeChunk (idx: number): Promise<void> {
return this.sendRequest({ method: 'closeChunk', params: [idx] })
}
server/backup/src/backup.ts
+57 -40
View File
@@ -134,6 +134,9 @@ export interface BackupInfo {
snapshots: BackupSnapshot[]
snapshotsIndex?: number
lastTxId?: string
// A hash of current domain transactions, so we could skip all other checks if same.
domainHashes: Record<Domain, string>
}
async function loadDigest (
@@ -143,15 +146,13 @@ async function loadDigest (
domain: Domain,
date?: number
): Promise<Map<Ref<Doc>, string>> {
ctx = ctx.newChild('load digest', { domain, count: snapshots.length })
ctx.info('loading-digest', { domain, snapshots: snapshots.length })
const result = new Map<Ref<Doc>, string>()
for (const s of snapshots) {
const d = s.domains[domain]
// Load old JSON snapshot
if (d?.snapshot !== undefined) {
const dChanges: SnapshotV6 = JSON.parse(gunzipSync((await storage.loadFile(d.snapshot)) as any).toString())
const dChanges: SnapshotV6 = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(d.snapshot))).toString())
for (const [k, v] of Object.entries(dChanges.added)) {
result.set(k as Ref<Doc>, v)
}
@@ -164,7 +165,7 @@ async function loadDigest (
}
for (const snapshot of d?.snapshots ?? []) {
try {
const dataBlob = gunzipSync((await storage.loadFile(snapshot)) as any)
const dataBlob = gunzipSync(new Uint8Array(await storage.loadFile(snapshot)))
.toString()
.split('\n')
const addedCount = parseInt(dataBlob.shift() ?? '0')
@@ -195,7 +196,6 @@ async function loadDigest (
break
}
}
ctx.end()
ctx.info('load-digest', { domain, snapshots: snapshots.length, documents: result.size })
return result
}
@@ -326,7 +326,7 @@ async function verifyDigest (
}
let lmodified = false
try {
const dataBlob = gunzipSync(await storage.loadFile(snapshot))
const dataBlob = gunzipSync(new Uint8Array(await storage.loadFile(snapshot)))
.toString()
.split('\n')
const addedCount = parseInt(dataBlob.shift() ?? '0')
@@ -389,7 +389,7 @@ async function write (chunk: any, stream: Writable): Promise<void> {
})
})
if (needDrain) {
await new Promise((resolve, reject) => stream.once('drain', resolve))
await new Promise((resolve) => stream.once('drain', resolve))
}
}
@@ -738,7 +738,8 @@ export async function backup (
let backupInfo: BackupInfo = {
workspace: workspaceId,
version: '0.6.2',
snapshots: []
snapshots: [],
domainHashes: {}
}
// Version 0.6.2, format of digest file is changed to
@@ -746,12 +747,17 @@ export async function backup (
const infoFile = 'backup.json.gz'
if (await storage.exists(infoFile)) {
backupInfo = JSON.parse(gunzipSync((await storage.loadFile(infoFile)) as any).toString())
backupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString())
}
backupInfo.version = '0.6.2'
backupInfo.workspace = workspaceId
if (backupInfo.domainHashes === undefined) {
// Migration
backupInfo.domainHashes = {}
}
let lastTx: Tx | undefined
let lastTxChecked = false
@@ -872,13 +878,6 @@ export async function backup (
let st = Date.now()
let changed: number = 0
const needRetrieveChunks: Ref<Doc>[][] = []
// Load all digest from collection.
ctx.info('processed', {
processed,
digest: digest.size,
time: Date.now() - st,
workspace: workspaceId
})
const oldHash = new Map<Ref<Doc>, string>()
function removeFromNeedRetrieve (needRetrieve: Ref<Doc>[], id: string): void {
@@ -961,7 +960,7 @@ export async function backup (
needRetrieve = []
}
if (currentChunk.finished) {
ctx.info('processed-end', {
ctx.info('processed', {
processed,
digest: digest.size,
time: Date.now() - st,
@@ -1015,6 +1014,11 @@ export async function backup (
removed: 0
}
const dHash = await connection.getDomainHash(domain)
if (backupInfo.domainHashes[domain] === dHash) {
ctx.info('no changes in domain', { domain })
return
}
// Cumulative digest
const digest = await ctx.with('load-digest', {}, (ctx) => loadDigest(ctx, storage, backupInfo.snapshots, domain))
@@ -1034,11 +1038,6 @@ export async function backup (
if (progress !== undefined) {
await progress(10)
}
if (needRetrieveChunks.length > 0) {
ctx.info('dumping domain...', { workspace: workspaceId, domain })
}
const totalChunks = needRetrieveChunks.flatMap((it) => it.length).reduce((p, c) => p + c, 0)
let processed = 0
let blobs = 0
@@ -1058,7 +1057,7 @@ export async function backup (
if (needRetrieve.length === 0) {
continue
}
ctx.info('Retrieve chunk', {
ctx.info('<<<< chunk', {
needRetrieve: needRetrieveChunks.reduce((v, docs) => v + docs.length, 0),
toLoad: needRetrieve.length,
workspace: workspaceId,
@@ -1066,7 +1065,7 @@ export async function backup (
})
let docs: Doc[] = []
try {
docs = await ctx.with('load-docs', {}, async (ctx) => await connection.loadDocs(domain, needRetrieve))
docs = await ctx.with('<<<< load-docs', {}, async () => await connection.loadDocs(domain, needRetrieve))
lastSize = docs.reduce((p, it) => p + estimateDocSize(it), 0)
if (docs.length !== needRetrieve.length) {
const nr = new Set(docs.map((it) => it._id))
@@ -1115,7 +1114,6 @@ export async function backup (
_pack = pack()
stIndex++
const storageFile = join(backupIndex, `${domain}-data-${snapshot.date}-${stIndex}.tar.gz`)
ctx.info('storing from domain', { domain, storageFile, workspace: workspaceId })
domainInfo.storage = [...(domainInfo.storage ?? []), storageFile]
const tmpFile = join(tmpRoot, basename(storageFile) + '.tmp')
const tempFile = createWriteStream(tmpFile)
@@ -1146,7 +1144,7 @@ export async function backup (
})
// We need to upload file to storage
ctx.info('Upload pack file', { storageFile, size: sz, workspace: workspaceId })
ctx.info('>>>> upload pack', { storageFile, size: sz, workspace: wsIds.url })
await storage.writeFile(storageFile, createReadStream(tmpFile))
await rm(tmpFile)
@@ -1283,7 +1281,10 @@ export async function backup (
changed++
}
if (changed > 0) {
if (changed > 0 || backupInfo.domainHashes[domain] !== dHash) {
// Store domain hash, to be used on next time.
backupInfo.domainHashes[domain] = dHash
snapshot.domains[domain] = domainInfo
domainInfo.added += processedChanges.added.size
domainInfo.updated += processedChanges.updated.size
@@ -1312,10 +1313,14 @@ export async function backup (
try {
global.gc?.()
} catch (err) {}
ctx.info('memory-stats', {
const mm = {
old: Math.round(oldUsed / (1024 * 1024)),
current: Math.round(process.memoryUsage().heapUsed / (1024 * 1024))
})
}
if (mm.old > mm.current + mm.current / 10) {
ctx.info('memory-stats', mm)
}
await ctx.with('process-domain', { domain }, async (ctx) => {
await processDomain(
ctx,
@@ -1338,7 +1343,7 @@ export async function backup (
let sizeInfo: Record<string, number> = {}
if (await storage.exists(sizeFile)) {
sizeInfo = JSON.parse(gunzipSync((await storage.loadFile(sizeFile)) as any).toString())
sizeInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(sizeFile))).toString())
}
let processed = 0
@@ -1436,7 +1441,7 @@ export async function backupList (storage: BackupStorage): Promise<void> {
if (!(await storage.exists(infoFile))) {
throw new Error(`${infoFile} should present to restore`)
}
const backupInfo: BackupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString())
const backupInfo: BackupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString())
console.log('workspace:', backupInfo.workspace ?? '', backupInfo.version)
for (const s of backupInfo.snapshots) {
console.log('snapshot: id:', s.date, ' date:', new Date(s.date))
@@ -1452,7 +1457,7 @@ export async function backupRemoveLast (storage: BackupStorage, date: number): P
if (!(await storage.exists(infoFile))) {
throw new Error(`${infoFile} should present to restore`)
}
const backupInfo: BackupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString())
const backupInfo: BackupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString())
console.log('workspace:', backupInfo.workspace ?? '', backupInfo.version)
const old = backupInfo.snapshots.length
backupInfo.snapshots = backupInfo.snapshots.filter((it) => it.date < date)
@@ -1474,7 +1479,7 @@ export async function backupSize (storage: BackupStorage): Promise<void> {
}
let size = 0
const backupInfo: BackupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString())
const backupInfo: BackupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString())
console.log('workspace:', backupInfo.workspace ?? '', backupInfo.version)
const addFileSize = async (file: string | undefined | null): Promise<void> => {
if (file != null && (await storage.exists(file))) {
@@ -1513,12 +1518,12 @@ export async function backupDownload (storage: BackupStorage, storeIn: string):
}
let size = 0
const backupInfo: BackupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString())
const backupInfo: BackupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString())
console.log('workspace:', backupInfo.workspace ?? '', backupInfo.version)
let sizeInfo: Record<string, number> = {}
if (await storage.exists(sizeFile)) {
sizeInfo = JSON.parse(gunzipSync(await storage.loadFile(sizeFile)).toString())
sizeInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(sizeFile))).toString())
}
console.log('workspace:', backupInfo.workspace ?? '', backupInfo.version)
@@ -1579,7 +1584,7 @@ export async function backupFind (storage: BackupStorage, id: Ref<Doc>, domain?:
if (!(await storage.exists(infoFile))) {
throw new Error(`${infoFile} should present to restore`)
}
const backupInfo: BackupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString())
const backupInfo: BackupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString())
console.log('workspace:', backupInfo.workspace ?? '', backupInfo.version)
const toolCtx = new MeasureMetricsContext('', {})
@@ -1676,7 +1681,7 @@ export async function restore (
ctx.error('file not pressent', { file: infoFile })
throw new Error(`${infoFile} should present to restore`)
}
const backupInfo: BackupInfo = JSON.parse(gunzipSync((await storage.loadFile(infoFile)) as any).toString())
const backupInfo: BackupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString())
let snapshots = backupInfo.snapshots
if (opt.date !== -1) {
const bk = backupInfo.snapshots.findIndex((it) => it.date === opt.date)
@@ -1688,6 +1693,10 @@ export async function restore (
} else {
opt.date = snapshots[snapshots.length - 1].date
}
if (backupInfo.domainHashes === undefined) {
backupInfo.domainHashes = {}
}
ctx.info('restore to ', { id: opt.date, date: new Date(opt.date).toDateString() })
const rsnapshots = Array.from(snapshots).reverse()
@@ -1764,6 +1773,11 @@ export async function restore (
}
async function processDomain (c: Domain): Promise<void> {
const dHash = await connection.getDomainHash(c)
if (backupInfo.domainHashes[c] === dHash) {
ctx.info('no changes in domain', { domain: c })
return
}
const changeset = await loadDigest(ctx, storage, snapshots, c, opt.date)
// We need to load full changeset from server
const serverChangeset = new Map<Ref<Doc>, string>()
@@ -1772,7 +1786,10 @@ export async function restore (
try {
global.gc?.()
} catch (err) {}
ctx.info('memory-stats', { old: oldUsed / (1024 * 1024), current: process.memoryUsage().heapUsed / (1024 * 1024) })
const mm = { old: oldUsed / (1024 * 1024), current: process.memoryUsage().heapUsed / (1024 * 1024) }
if (mm.old > mm.current + mm.current / 10) {
ctx.info('memory-stats', mm)
}
let idx: number | undefined
let loaded = 0
@@ -2177,7 +2194,7 @@ export async function compactBackup (
const infoFile = 'backup.json.gz'
if (await storage.exists(infoFile)) {
backupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString())
backupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString())
} else {
console.log('No backup found')
return
@@ -2514,7 +2531,7 @@ export async function checkBackupIntegrity (ctx: MeasureContext, storage: Backup
const infoFile = 'backup.json.gz'
if (await storage.exists(infoFile)) {
backupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString())
backupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString())
} else {
console.log('No backup found')
return
server/backup/src/service.ts
+29 -21
View File
@@ -140,7 +140,7 @@ class BackupWorker {
}
const lastBackup = it.backupInfo?.lastBackup ?? 0
if ((now - lastBackup) / 1000 < this.config.Interval) {
if ((now - lastBackup) / 1000 < this.config.Interval && this.config.Interval !== 0) {
// No backup required, interval not elapsed
skipped++
return false
@@ -187,7 +187,7 @@ class BackupWorker {
ctx.warn('Preparing for BACKUP', {
total: workspaces.length,
skipped,
workspaces: workspaces.map((it) => it.uuid)
workspaces: workspaces.map((it) => it.url)
})
const part = workspaces.slice(0, 500)
@@ -222,26 +222,34 @@ class BackupWorker {
ETA: Math.round((workspaces.length - processed) * avgTime)
})
}, 10000)
for (const ws of workspaces) {
await rateLimiter.add(async () => {
index++
if (this.canceled || Date.now() - startTime > recheckTimeout) {
return // If canceled, we should stop
}
const st = Date.now()
const result = await this.doBackup(ctx, ws)
const totalTime = Date.now() - st
times.push(totalTime)
if (!result) {
failedWorkspaces.push(ws)
return
}
processed++
})
}
await rateLimiter.waitProcessing()
clearInterval(infoTo)
try {
for (const ws of workspaces) {
await rateLimiter.add(async () => {
try {
index++
if (this.canceled || Date.now() - startTime > recheckTimeout) {
return // If canceled, we should stop
}
const st = Date.now()
const result = await this.doBackup(ctx, ws)
const totalTime = Date.now() - st
times.push(totalTime)
if (!result) {
failedWorkspaces.push(ws)
return
}
processed++
} catch (err: any) {
ctx.error('Backup failed', { err })
}
})
}
await rateLimiter.waitProcessing()
} finally {
clearInterval(infoTo)
}
return { failedWorkspaces, processed, skipped: workspaces.length - processed }
}
server/core/src/benchmark/index.ts
+6
View File
@@ -22,6 +22,7 @@ import core, {
type Class,
type Doc,
type DocumentQuery,
type Domain,
type FindOptions,
type FindResult,
type Hierarchy,
@@ -90,6 +91,11 @@ class BenchmarkDbAdapter extends DummyDbAdapter {
return toFindResult<T>(result as T[])
}
getDomainHash (ctx: MeasureContext, domain: Domain): Promise<string> {
// Since benchmark coult not be changed.
return Promise.resolve('')
}
tx (ctx: MeasureContext, ...tx: Tx[]): Promise<TxResult[]> {
if (benchData === '') {
benchData = genData(1024 * 1024)
server/core/src/mem.ts
+5 -1
View File
@@ -37,7 +37,6 @@ import core, {
type WorkspaceIds
} from '@hcengineering/core'
import { type DbAdapter, type DbAdapterHandler, type DomainHelperOperations } from './adapter'
/**
* @public
*/
@@ -102,6 +101,11 @@ export class DummyDbAdapter implements DbAdapter {
async clean (ctx: MeasureContext, domain: Domain, docs: Ref<Doc>[]): Promise<void> {}
getDomainHash (ctx: MeasureContext, domain: Domain): Promise<string> {
// Return '' for empty documents content.
return Promise.resolve('')
}
async update<T extends Doc>(
ctx: MeasureContext,
domain: Domain,
server/core/src/storage.ts
+4
View File
@@ -88,6 +88,10 @@ export class BackupClientOps {
})
}
getDomainHash (ctx: MeasureContext, domain: Domain): Promise<string> {
return this.storage.getDomainHash(ctx, domain)
}
closeChunk (ctx: MeasureContext, idx: number): Promise<void> {
return ctx.with('close-chunk', {}, async () => {
const chunk = this.chunkInfo.get(idx)
server/core/src/types.ts
+2
View File
@@ -572,6 +572,8 @@ export interface Session {
searchFulltext: (ctx: ClientSessionCtx, query: SearchQuery, options: SearchOptions) => Promise<void>
tx: (ctx: ClientSessionCtx, tx: Tx) => Promise<void>
loadChunk: (ctx: ClientSessionCtx, domain: Domain, idx?: number) => Promise<void>
getDomainHash: (ctx: ClientSessionCtx, domain: Domain) => Promise<void>
closeChunk: (ctx: ClientSessionCtx, idx: number) => Promise<void>
loadDocs: (ctx: ClientSessionCtx, domain: Domain, docs: Ref<Doc>[]) => Promise<void>
upload: (ctx: ClientSessionCtx, domain: Domain, docs: Doc[]) => Promise<void>
server/core/src/utils.ts
+34 -1
View File
@@ -31,7 +31,7 @@ import core, {
type PersonUuid
} from '@hcengineering/core'
import { PlatformError, unknownError } from '@hcengineering/platform'
import { type Hash } from 'crypto'
import { createHash, type Hash } from 'crypto'
import fs from 'fs'
import type { DbAdapter } from './adapter'
import { BackupClientOps } from './storage'
@@ -262,6 +262,7 @@ export function wrapPipeline (ctx: MeasureContext, pipeline: Pipeline, wsIds: Wo
getHierarchy: () => pipeline.context.hierarchy,
getModel: () => pipeline.context.modelDb,
loadChunk: (domain, idx) => backupOps.loadChunk(ctx, domain, idx),
getDomainHash: (domain) => backupOps.getDomainHash(ctx, domain),
loadDocs: (domain, docs) => backupOps.loadDocs(ctx, domain, docs),
upload: (domain, docs) => backupOps.upload(ctx, domain, docs),
searchFulltext: async (query, options) => ({ docs: [], total: 0 }),
@@ -310,6 +311,10 @@ export function wrapAdapterToClient (ctx: MeasureContext, storageAdapter: DbAdap
throw new Error('unsupported')
}
async getDomainHash (domain: Domain): Promise<string> {
return await storageAdapter.getDomainHash(ctx, domain)
}
async closeChunk (idx: number): Promise<void> {}
async loadDocs (domain: Domain, docs: Ref<Doc>[]): Promise<Doc[]> {
@@ -328,3 +333,31 @@ export function wrapAdapterToClient (ctx: MeasureContext, storageAdapter: DbAdap
}
return new TestClientConnection()
}
export async function calcHashHash (ctx: MeasureContext, domain: Domain, adapter: DbAdapter): Promise<string> {
const hash = createHash('sha256')
const it = adapter.find(ctx, domain)
try {
let count = 0
while (true) {
const part = await it.next(ctx)
if (part.length === 0) {
break
}
count += part.length
for (const doc of part) {
hash.update(doc.id)
hash.update(doc.hash)
}
}
if (count === 0) {
// Use empty hash for empty documents.
return ''
}
return hash.digest('hex')
} finally {
await it.close(ctx)
}
}
server/middleware/src/lowLevel.ts
+3
View File
@@ -75,6 +75,9 @@ export class LowLevelMiddleware extends BaseMiddleware implements Middleware {
rawDeleteMany (domain, query) {
return adapterManager.getAdapter(domain, true).rawDeleteMany(domain, query)
},
getDomainHash (ctx: MeasureContext, domain: Domain): Promise<string> {
return adapterManager.getAdapter(domain, false).getDomainHash(ctx, domain)
},
traverse<T extends Doc>(
domain: Domain,
query: DocumentQuery<T>,
server/mongo/src/storage.ts
+7 -1
View File
@@ -69,7 +69,8 @@ import {
type DomainHelperOperations,
type ServerFindOptions,
type StorageAdapter,
type TxAdapter
type TxAdapter,
calcHashHash
} from '@hcengineering/server-core'
import {
type AbstractCursor,
@@ -1084,6 +1085,11 @@ abstract class MongoAdapterBase implements DbAdapter {
return Date.now().toString(16) // Current hash value
}
@withContext('get-domain-hash')
async getDomainHash (ctx: MeasureContext, domain: Domain): Promise<string> {
return await calcHashHash(ctx, domain, this)
}
strimSize (str?: string): string {
if (str == null) {
return ''
server/postgres/src/storage.ts
+7 -1
View File
@@ -65,7 +65,8 @@ import {
type DbAdapterHandler,
type DomainHelperOperations,
type ServerFindOptions,
type TxAdapter
type TxAdapter,
calcHashHash
} from '@hcengineering/server-core'
import type postgres from 'postgres'
import { createDBClient, createGreenDBClient, type DBClient } from './client'
@@ -1439,6 +1440,11 @@ abstract class PostgresAdapterBase implements DbAdapter {
return res
}
@withContext('get-domain-hash')
async getDomainHash (ctx: MeasureContext, domain: Domain): Promise<string> {
return await calcHashHash(ctx, domain, this)
}
curHash (): string {
return Date.now().toString(16) // Current hash value
}
server/server-pipeline/src/blobStorage.ts
+7 -1
View File
@@ -33,7 +33,8 @@ import {
type TxResult,
type Blob,
type WorkspaceIds,
type WorkspaceDataId
type WorkspaceDataId,
generateId
} from '@hcengineering/core'
import { PlatformError, unknownError } from '@hcengineering/platform'
import {
@@ -123,6 +124,11 @@ class StorageBlobAdapter implements DbAdapter {
return blobs
}
getDomainHash (ctx: MeasureContext, domain: Domain): Promise<string> {
// TODO: Check if possible to ask storage if there any changes.
return Promise.resolve(generateId())
}
async upload (ctx: MeasureContext, domain: Domain, docs: Doc[]): Promise<void> {
// Nothing to do
}
server/server/src/client.ts
+11
View File
@@ -250,6 +250,17 @@ export class ClientSession implements Session {
}
}
async getDomainHash (ctx: ClientSessionCtx, domain: Domain): Promise<void> {
this.lastRequest = Date.now()
try {
const result = await this.getOps(ctx.pipeline).getDomainHash(ctx.ctx, domain)
await ctx.sendResponse(ctx.requestId, result)
} catch (err: any) {
await ctx.sendError(ctx.requestId, 'Failed to upload', unknownError(err))
ctx.ctx.error('failed to getDomainHash', { domain, err })
}
}
async closeChunk (ctx: ClientSessionCtx, idx: number): Promise<void> {
this.lastRequest = Date.now()
await this.getOps(ctx.pipeline).closeChunk(ctx.ctx, idx)