mirror of
https://github.com/hcengineering/platform.git
synced 2026-08-21 20:02:22 +02:00
1955 lines
64 KiB
TypeScript
1955 lines
64 KiB
TypeScript
/* eslint-disable @typescript-eslint/no-unused-vars */
|
|
import {
|
|
type AccountDB,
|
|
type MongoAccountDB,
|
|
type Workspace,
|
|
addSocialIdToPerson,
|
|
ensurePerson,
|
|
findFullSocialIdBySocialKey,
|
|
findPersonBySocialKey,
|
|
mergeSpecifiedPersons,
|
|
mergeSpecifiedAccounts,
|
|
createAccount
|
|
} from '@hcengineering/account'
|
|
import { getFirstName, getLastName } from '@hcengineering/contact'
|
|
import {
|
|
systemAccountUuid,
|
|
type BackupClient,
|
|
type Client,
|
|
type Doc,
|
|
MeasureMetricsContext,
|
|
SocialIdType,
|
|
type PersonUuid,
|
|
type SocialKey,
|
|
type AccountUuid,
|
|
parseSocialIdString,
|
|
DOMAIN_SPACE,
|
|
AccountRole,
|
|
generateId,
|
|
type WorkspaceDataId,
|
|
type WorkspaceUuid,
|
|
generateUuid,
|
|
type WorkspaceMode,
|
|
type Tx
|
|
} from '@hcengineering/core'
|
|
import { getMongoClient, getWorkspaceMongoDB } from '@hcengineering/mongo'
|
|
import {
|
|
convertDoc,
|
|
createTables,
|
|
getDBClient,
|
|
getDocFieldsByDomains,
|
|
retryTxn,
|
|
translateDomain,
|
|
setDBExtraOptions
|
|
} from '@hcengineering/postgres'
|
|
import { type DBDoc } from '@hcengineering/postgres/types/utils'
|
|
import { getTransactorEndpoint } from '@hcengineering/server-client'
|
|
import { generateToken } from '@hcengineering/server-token'
|
|
import { connect, sendTransactorEvent } from '@hcengineering/server-tool'
|
|
import {
|
|
type MongoAccountDB as v6MongoAccountDB,
|
|
type Account as OldAccount,
|
|
type Workspace as OldWorkspace
|
|
} from '@hcengineering/account-service'
|
|
import { type MongoClient } from 'mongodb'
|
|
import type postgres from 'postgres'
|
|
import { type Row } from 'postgres'
|
|
import { getToolToken } from './utils'
|
|
import { type BackupStorage, createFileBackupStorage, restore } from '@hcengineering/server-backup'
|
|
import { buildStorageFromConfig, storageConfigFromEnv } from '@hcengineering/server-storage'
|
|
import { getPlatformQueue } from '@hcengineering/kafka'
|
|
import {
|
|
type Pipeline,
|
|
QueueTopic,
|
|
type QueueWorkspaceMessage,
|
|
type StorageAdapter,
|
|
workspaceEvents
|
|
} from '@hcengineering/server-core'
|
|
import { createBackupPipeline, createEmptyBroadcastOps } from '@hcengineering/server-pipeline'
|
|
|
|
export async function moveFromMongoToPG (
|
|
accountDb: AccountDB,
|
|
mongoUrl: string,
|
|
dbUrl: string | undefined,
|
|
workspaces: Workspace[],
|
|
region: string
|
|
): Promise<void> {
|
|
if (dbUrl === undefined) {
|
|
throw new Error('dbUrl is required')
|
|
}
|
|
const client = getMongoClient(mongoUrl)
|
|
const mongo = await client.getClient()
|
|
const pg = getDBClient(dbUrl)
|
|
const pgClient = await pg.getClient()
|
|
|
|
for (let index = 0; index < workspaces.length; index++) {
|
|
const ws = workspaces[index]
|
|
try {
|
|
await moveWorkspace(accountDb, mongo, pgClient, ws, region)
|
|
console.log('Move workspace', index, workspaces.length)
|
|
} catch (err) {
|
|
console.log('Error when move workspace', ws.name ?? ws.url, err)
|
|
throw err
|
|
}
|
|
}
|
|
pg.close()
|
|
client.close()
|
|
}
|
|
|
|
async function moveWorkspace (
|
|
accountDb: AccountDB,
|
|
mongo: MongoClient,
|
|
pgClient: postgres.Sql,
|
|
ws: Workspace,
|
|
region: string,
|
|
include?: Set<string>,
|
|
force = false
|
|
): Promise<void> {
|
|
try {
|
|
console.log('move workspace', ws.name ?? ws.url)
|
|
const wsId = ws.uuid
|
|
// TODO: get workspace mongoDB
|
|
const mongoDB = getWorkspaceMongoDB(mongo, ws.dataId ?? wsId)
|
|
const collections = await mongoDB.collections()
|
|
let tables = collections.map((c) => c.collectionName)
|
|
if (include !== undefined) {
|
|
tables = tables.filter((t) => include.has(t))
|
|
}
|
|
|
|
await createTables(new MeasureMetricsContext('', {}), pgClient, '', tables)
|
|
const token = generateToken(systemAccountUuid, wsId, { service: 'tool' })
|
|
const endpoint = await getTransactorEndpoint(token, 'external')
|
|
const connection = (await connect(endpoint, wsId, undefined, {
|
|
model: 'upgrade'
|
|
})) as unknown as Client & BackupClient
|
|
for (const collection of collections) {
|
|
const domain = translateDomain(collection.collectionName)
|
|
if (include !== undefined && !include.has(domain)) {
|
|
continue
|
|
}
|
|
const cursor = collection.find()
|
|
const current = await pgClient`SELECT _id FROM ${pgClient(domain)} WHERE "workspaceId" = ${ws.uuid}`
|
|
const currentIds = new Set(current.map((r) => r._id))
|
|
console.log('move domain', domain)
|
|
const docs: Doc[] = []
|
|
const fields = getDocFieldsByDomains(domain)
|
|
const filedsWithData = [...fields, 'data']
|
|
const insertFields: string[] = ['workspaceId']
|
|
for (const field of filedsWithData) {
|
|
insertFields.push(field)
|
|
}
|
|
while (true) {
|
|
const toRemove: string[] = []
|
|
while (docs.length < 5000) {
|
|
const doc = (await cursor.next()) as Doc | null
|
|
if (doc === null) break
|
|
if (currentIds.has(doc._id)) {
|
|
if (force) {
|
|
toRemove.push(doc._id)
|
|
} else {
|
|
continue
|
|
}
|
|
}
|
|
docs.push(doc)
|
|
}
|
|
while (toRemove.length > 0) {
|
|
const part = toRemove.splice(0, 100)
|
|
await retryTxn(pgClient, async (client) => {
|
|
await client.unsafe(
|
|
`DELETE FROM ${translateDomain(domain)} WHERE "workspaceId" = '${ws.uuid}' AND _id IN (${part.map((c) => `'${c}'`).join(', ')})`
|
|
)
|
|
})
|
|
}
|
|
if (docs.length === 0) break
|
|
while (docs.length > 0) {
|
|
const part = docs.splice(0, 100)
|
|
const values: DBDoc[] = []
|
|
for (let i = 0; i < part.length; i++) {
|
|
const doc = part[i]
|
|
const d = convertDoc(domain, doc, wsId)
|
|
values.push(d)
|
|
}
|
|
try {
|
|
await retryTxn(pgClient, async (client) => {
|
|
await client`INSERT INTO ${client(translateDomain(domain))} ${client(values, insertFields)}`
|
|
})
|
|
} catch (err) {
|
|
console.log('Error when insert', domain, err)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
// TODO: FIXME
|
|
// await updateWorkspace(accountDb, ws, { region })
|
|
await connection.sendForceClose()
|
|
await connection.close()
|
|
} catch (err) {
|
|
console.log('Error when move workspace', ws.name ?? ws.url, err)
|
|
throw err
|
|
}
|
|
}
|
|
|
|
export async function moveWorkspaceFromMongoToPG (
|
|
accountDb: AccountDB,
|
|
mongoUrl: string,
|
|
dbUrl: string | undefined,
|
|
ws: Workspace,
|
|
region: string,
|
|
include?: Set<string>,
|
|
force?: boolean
|
|
): Promise<void> {
|
|
if (dbUrl === undefined) {
|
|
throw new Error('dbUrl is required')
|
|
}
|
|
const client = getMongoClient(mongoUrl)
|
|
const mongo = await client.getClient()
|
|
const pg = getDBClient(dbUrl)
|
|
const pgClient = await pg.getClient()
|
|
|
|
await moveWorkspace(accountDb, mongo, pgClient, ws, region, include, force)
|
|
pg.close()
|
|
client.close()
|
|
}
|
|
|
|
export async function moveAccountDbFromMongoToPG (
|
|
ctx: MeasureMetricsContext,
|
|
mongoDb: AccountDB,
|
|
pgDb: AccountDB
|
|
): Promise<void> {
|
|
const mdb = mongoDb as MongoAccountDB
|
|
const BATCH_SIZE = 5000
|
|
const WS_BATCH_SIZE = 2000
|
|
|
|
ctx.info('Starting migration of persons...')
|
|
const personsCursor = mdb.person.findCursor({})
|
|
try {
|
|
let personsCount = 0
|
|
let personsBatch: any[] = []
|
|
|
|
while (await personsCursor.hasNext()) {
|
|
const person = await personsCursor.next()
|
|
if (person == null) break
|
|
|
|
const exists = await pgDb.person.findOne({ uuid: person.uuid })
|
|
if (exists == null) {
|
|
if (person.firstName == null) {
|
|
person.firstName = 'n/a'
|
|
}
|
|
|
|
if (person.lastName == null) {
|
|
person.lastName = ''
|
|
}
|
|
|
|
personsBatch.push(person)
|
|
if (personsBatch.length >= BATCH_SIZE) {
|
|
await pgDb.person.insertMany(personsBatch)
|
|
personsCount += personsBatch.length
|
|
ctx.info(`Migrated ${personsCount} persons...`)
|
|
personsBatch = []
|
|
}
|
|
}
|
|
}
|
|
// Insert remaining batch
|
|
if (personsBatch.length > 0) {
|
|
await pgDb.person.insertMany(personsBatch)
|
|
personsCount += personsBatch.length
|
|
}
|
|
ctx.info(`Migrated ${personsCount} persons`)
|
|
} finally {
|
|
await personsCursor.close()
|
|
}
|
|
|
|
ctx.info('Starting migration of accounts...')
|
|
const accountsCursor = mdb.account.findCursor({})
|
|
try {
|
|
let accountsCount = 0
|
|
let accountsBatch: any[] = []
|
|
let passwordsBatch: any[] = []
|
|
|
|
while (await accountsCursor.hasNext()) {
|
|
const account = await accountsCursor.next()
|
|
if (account == null) break
|
|
|
|
const exists = await pgDb.account.findOne({ uuid: account.uuid })
|
|
if (exists == null) {
|
|
const { hash, salt } = account
|
|
|
|
delete account.hash
|
|
delete account.salt
|
|
|
|
accountsBatch.push(account)
|
|
if (hash != null && salt != null) {
|
|
passwordsBatch.push([account.uuid, hash, salt])
|
|
}
|
|
|
|
if (accountsBatch.length >= BATCH_SIZE) {
|
|
await pgDb.account.insertMany(accountsBatch)
|
|
for (const [accountUuid, hash, salt] of passwordsBatch) {
|
|
await pgDb.setPassword(accountUuid, hash, salt)
|
|
}
|
|
|
|
accountsCount += accountsBatch.length
|
|
ctx.info(`Migrated ${accountsCount} accounts...`)
|
|
accountsBatch = []
|
|
passwordsBatch = []
|
|
}
|
|
}
|
|
}
|
|
// Insert remaining batch
|
|
if (accountsBatch.length > 0) {
|
|
await pgDb.account.insertMany(accountsBatch)
|
|
accountsCount += accountsBatch.length
|
|
}
|
|
if (passwordsBatch.length > 0) {
|
|
for (const [accountUuid, hash, salt] of passwordsBatch) {
|
|
await pgDb.setPassword(accountUuid, hash, salt)
|
|
}
|
|
}
|
|
ctx.info(`Migrated ${accountsCount} accounts`)
|
|
} finally {
|
|
await accountsCursor.close()
|
|
}
|
|
|
|
ctx.info('Starting migration of social IDs...')
|
|
const socialIdsCursor = mdb.socialId.findCursor({})
|
|
try {
|
|
let socialIdsCount = 0
|
|
let socialIdsBatch: any[] = []
|
|
while (await socialIdsCursor.hasNext()) {
|
|
const socialId = await socialIdsCursor.next()
|
|
if (socialId == null) break
|
|
|
|
const exists = await pgDb.socialId.findOne({ key: socialId.key })
|
|
if (exists == null) {
|
|
delete (socialId as any).key
|
|
delete (socialId as any).id
|
|
delete (socialId as any)._id // Types of _id are incompatible
|
|
|
|
socialIdsBatch.push(socialId)
|
|
|
|
if (socialIdsBatch.length >= BATCH_SIZE) {
|
|
await pgDb.socialId.insertMany(socialIdsBatch)
|
|
|
|
socialIdsCount += socialIdsBatch.length
|
|
ctx.info(`Migrated ${socialIdsCount} social ids...`)
|
|
socialIdsBatch = []
|
|
}
|
|
}
|
|
}
|
|
// Insert remaining batch
|
|
if (socialIdsBatch.length > 0) {
|
|
await pgDb.socialId.insertMany(socialIdsBatch)
|
|
socialIdsCount += socialIdsBatch.length
|
|
}
|
|
ctx.info(`Migrated ${socialIdsCount} social IDs`)
|
|
} finally {
|
|
await socialIdsCursor.close()
|
|
}
|
|
|
|
ctx.info('Starting migration of account events...')
|
|
const accountEventsCursor = mdb.accountEvent.findCursor({})
|
|
try {
|
|
let eventsCount = 0
|
|
let eventsBatch: any[] = []
|
|
while (await accountEventsCursor.hasNext()) {
|
|
const accountEvent = await accountEventsCursor.next()
|
|
if (accountEvent == null) break
|
|
|
|
const exists = await pgDb.accountEvent.findOne({
|
|
accountUuid: accountEvent.accountUuid,
|
|
eventType: accountEvent.eventType,
|
|
time: accountEvent.time
|
|
})
|
|
if (exists == null) {
|
|
const account = await pgDb.account.findOne({ uuid: accountEvent.accountUuid })
|
|
if (account == null) continue // Not a big deal if we don't move the event for non-existing account
|
|
|
|
eventsBatch.push(accountEvent)
|
|
|
|
if (eventsBatch.length >= BATCH_SIZE) {
|
|
await pgDb.accountEvent.insertMany(eventsBatch)
|
|
eventsCount += eventsBatch.length
|
|
ctx.info(`Migrated ${eventsCount} account events...`)
|
|
eventsBatch = []
|
|
}
|
|
}
|
|
}
|
|
// Insert remaining batch
|
|
if (eventsBatch.length > 0) {
|
|
await pgDb.accountEvent.insertMany(eventsBatch)
|
|
eventsCount += eventsBatch.length
|
|
}
|
|
ctx.info(`Migrated ${eventsCount} account events`)
|
|
} finally {
|
|
await accountEventsCursor.close()
|
|
}
|
|
|
|
ctx.info('Starting migration of workspaces...')
|
|
const workspacesCursor = mdb.workspace.findCursor({})
|
|
try {
|
|
let workspacesCount = 0
|
|
let workspacesBatch: any[] = []
|
|
let workspacesStatusesBatch: any[] = []
|
|
let workspacesMembersBatch: any[] = []
|
|
let membersCount = 0
|
|
while (await workspacesCursor.hasNext()) {
|
|
const workspace = await workspacesCursor.next()
|
|
if (workspace == null) break
|
|
|
|
const exists = await pgDb.workspace.findOne({ uuid: workspace.uuid })
|
|
if (exists != null) continue
|
|
|
|
const status = workspace.status
|
|
if (status == null) continue
|
|
delete (workspace as any).status
|
|
|
|
if (workspace.createdBy === 'N/A') {
|
|
delete workspace.createdBy
|
|
}
|
|
|
|
if (workspace.billingAccount === 'N/A') {
|
|
delete workspace.billingAccount
|
|
}
|
|
|
|
if (workspace.createdOn == null) {
|
|
workspace.createdOn = Date.now()
|
|
}
|
|
|
|
workspacesBatch.push(workspace)
|
|
workspacesStatusesBatch.push(status)
|
|
|
|
const members = await mdb.getWorkspaceMembers(workspace.uuid)
|
|
for (const member of members) {
|
|
const alreadyAssigned = await pgDb.getWorkspaceRole(member.person, workspace.uuid)
|
|
if (alreadyAssigned != null) continue
|
|
|
|
workspacesMembersBatch.push([member.person, workspace.uuid, member.role])
|
|
}
|
|
|
|
if (workspacesBatch.length >= WS_BATCH_SIZE) {
|
|
const workspaceUuids = await pgDb.workspace.insertMany(workspacesBatch)
|
|
workspacesCount += workspacesBatch.length
|
|
workspacesBatch = []
|
|
|
|
await pgDb.workspaceStatus.insertMany(
|
|
workspacesStatusesBatch.map((s, i) => ({ ...s, workspaceUuid: workspaceUuids[i] }))
|
|
)
|
|
workspacesStatusesBatch = []
|
|
|
|
await pgDb.batchAssignWorkspace(workspacesMembersBatch)
|
|
membersCount += workspacesMembersBatch.length
|
|
workspacesMembersBatch = []
|
|
|
|
ctx.info(`Migrated ${workspacesCount} workspaces...`)
|
|
}
|
|
}
|
|
|
|
// Insert remaining batch
|
|
if (workspacesBatch.length > 0) {
|
|
const workspaceUuids = await pgDb.workspace.insertMany(workspacesBatch)
|
|
workspacesCount += workspacesBatch.length
|
|
await pgDb.workspaceStatus.insertMany(
|
|
workspacesStatusesBatch.map((s, i) => ({ ...s, workspaceUuid: workspaceUuids[i] }))
|
|
)
|
|
await pgDb.batchAssignWorkspace(workspacesMembersBatch)
|
|
membersCount += workspacesMembersBatch.length
|
|
}
|
|
|
|
ctx.info(`Migrated ${workspacesCount} workspaces with ${membersCount} member assignments`)
|
|
} finally {
|
|
await workspacesCursor.close()
|
|
}
|
|
|
|
ctx.info('Starting migration of invites...')
|
|
const invitesCursor = mdb.invite.findCursor({})
|
|
try {
|
|
// eslint-disable-next-line @typescript-eslint/no-loss-of-precision
|
|
const MAX_INT_8 = 9223372036854775807
|
|
let invitesCount = 0
|
|
let invitesBatch: any[] = []
|
|
while (await invitesCursor.hasNext()) {
|
|
const invite = await invitesCursor.next()
|
|
if (invite == null) break
|
|
if (invite.migratedFrom == null) {
|
|
invite.migratedFrom = invite.id
|
|
}
|
|
|
|
delete (invite as any).id
|
|
|
|
if (invite.expiresOn > MAX_INT_8 || typeof invite.expiresOn !== 'number') {
|
|
invite.expiresOn = -1
|
|
}
|
|
|
|
if (["USER'", 'ADMIN'].includes(invite.role as any)) {
|
|
invite.role = AccountRole.User
|
|
}
|
|
|
|
const exists = await pgDb.invite.findOne({ migratedFrom: invite.migratedFrom })
|
|
if (exists == null) {
|
|
invitesBatch.push(invite)
|
|
|
|
if (invitesBatch.length >= BATCH_SIZE) {
|
|
await pgDb.invite.insertMany(invitesBatch)
|
|
invitesCount += invitesBatch.length
|
|
ctx.info(`Migrated ${invitesCount} invites...`)
|
|
invitesBatch = []
|
|
}
|
|
}
|
|
}
|
|
if (invitesBatch.length > 0) {
|
|
await pgDb.invite.insertMany(invitesBatch)
|
|
invitesCount += invitesBatch.length
|
|
}
|
|
ctx.info(`Migrated ${invitesCount} invites`)
|
|
} finally {
|
|
await invitesCursor.close()
|
|
}
|
|
|
|
ctx.info('Account database migration completed')
|
|
}
|
|
|
|
export async function migrateCreatedModifiedBy (
|
|
ctx: MeasureMetricsContext,
|
|
dbUrl: string,
|
|
workspace: Workspace,
|
|
includeDomains?: string[],
|
|
excludeDomains?: string[],
|
|
maxLifetimeSec?: number,
|
|
batchSize?: number,
|
|
force: boolean = false,
|
|
maxReconnects: number = 30,
|
|
maxRetries: number = 50
|
|
): Promise<void> {
|
|
if (!dbUrl.startsWith('postgresql')) {
|
|
throw new Error('Only CockroachDB is supported')
|
|
}
|
|
|
|
const wsUuid = workspace.uuid
|
|
ctx.info('Processing workspace', {
|
|
workspaceUuid: workspace.uuid,
|
|
workspaceName: workspace.name,
|
|
workspaceUrl: workspace.url
|
|
})
|
|
|
|
if (maxLifetimeSec !== undefined) {
|
|
setDBExtraOptions({ max_lifetime: maxLifetimeSec })
|
|
}
|
|
|
|
let progressMade = false
|
|
let connectsCount = 0
|
|
let retriesCount = 0
|
|
let reconnecting = false
|
|
let retrying = false
|
|
let done = false
|
|
let pg: ReturnType<typeof getDBClient> | undefined
|
|
let pgClient: postgres.Sql | undefined
|
|
|
|
while (!done && (connectsCount === 0 || retrying || (reconnecting && progressMade))) {
|
|
try {
|
|
if (connectsCount === 0 || reconnecting) {
|
|
ctx.info(reconnecting ? ' Reconnecting...' : ' Connecting...')
|
|
|
|
progressMade = false
|
|
connectsCount++
|
|
|
|
pg = getDBClient(dbUrl)
|
|
pgClient = await pg.getClient()
|
|
|
|
// Expect temp table with mapping to be created manually
|
|
// Create mapping table
|
|
await pgClient`
|
|
CREATE TABLE IF NOT EXISTS temp_data.account_personid_mapping_v2 (
|
|
workspace_id uuid,
|
|
old_account_id text,
|
|
new_person_id text,
|
|
CONSTRAINT account_personid_mapping_v2_pk PRIMARY KEY (workspace_id, old_account_id)
|
|
)
|
|
`
|
|
|
|
const [res] = await pgClient`SELECT COUNT(*) FROM temp_data.account_personid_mapping_v2`
|
|
|
|
if (res.count === '0') {
|
|
// Populate mapping table
|
|
await pgClient`
|
|
INSERT INTO temp_data.account_personid_mapping_v2
|
|
WITH accounts as (
|
|
SELECT
|
|
tx."workspaceId" as workspace_id,
|
|
tx."objectId" as old_account_id,
|
|
COALESCE(
|
|
-- Get the latest email from updates
|
|
(
|
|
SELECT tx2.data->'operations'->>'email'
|
|
FROM model_tx tx2
|
|
WHERE tx2."objectId" = tx."objectId"
|
|
AND tx2."workspaceId" = tx."workspaceId"
|
|
AND tx2.data->>'objectClass' = 'contact:class:PersonAccount'
|
|
AND tx2.data->'operations'->>'email' IS NOT NULL
|
|
ORDER BY tx2."createdOn" DESC
|
|
LIMIT 1
|
|
),
|
|
-- If no updates with email, get from create transaction
|
|
tx.data->'attributes'->>'email'
|
|
) as latest_email
|
|
FROM model_tx tx
|
|
WHERE tx."_class" = 'core:class:TxCreateDoc'
|
|
AND tx.data->>'objectClass' = 'contact:class:PersonAccount'
|
|
AND tx.data->'attributes'->>'email' IS NOT null
|
|
),
|
|
account_data as (
|
|
SELECT
|
|
workspace_id,
|
|
old_account_id,
|
|
CASE
|
|
WHEN latest_email LIKE 'github:%' THEN lower(latest_email)
|
|
WHEN latest_email LIKE 'openid:%' THEN 'oidc:' || lower(substring(latest_email from 8))
|
|
ELSE 'email:' || lower(latest_email)
|
|
END as social_key
|
|
FROM accounts
|
|
WHERE latest_email IS NOT NULL AND latest_email != ''
|
|
)
|
|
SELECT
|
|
ad.workspace_id,
|
|
ad.old_account_id,
|
|
si."_id" as new_person_id
|
|
FROM account_data ad
|
|
JOIN global_account.social_id si ON si."key" = ad.social_key
|
|
WHERE ad.old_account_id NOT IN ('core:account:System', 'core:account:ConfigUser')
|
|
`
|
|
}
|
|
|
|
// Create progress table
|
|
await pgClient`
|
|
CREATE TABLE IF NOT EXISTS temp_data.account_personid_mapping_v2_progress (
|
|
workspace_id text,
|
|
domain text,
|
|
field text,
|
|
CONSTRAINT account_personid_mapping_v2_progress_pk PRIMARY KEY (workspace_id, domain, field)
|
|
)
|
|
`
|
|
}
|
|
|
|
if (pgClient == null) {
|
|
throw new Error('Could not connect to postgres')
|
|
}
|
|
|
|
if (retrying) {
|
|
retriesCount++
|
|
}
|
|
|
|
reconnecting = false
|
|
retrying = false
|
|
|
|
// Get list of tables to process
|
|
const tables = await pgClient`
|
|
SELECT table_name
|
|
FROM information_schema.columns
|
|
WHERE table_schema = 'public'
|
|
AND column_name IN ('createdBy', 'modifiedBy')
|
|
GROUP BY table_name
|
|
HAVING COUNT(DISTINCT column_name) = 2
|
|
`
|
|
let filteredTables: Row[] = tables
|
|
if (includeDomains != null && includeDomains.length > 0) {
|
|
filteredTables = tables.filter((t) => includeDomains.includes(t.table_name))
|
|
}
|
|
if (excludeDomains != null && excludeDomains.length > 0) {
|
|
filteredTables = filteredTables.filter((t) => !excludeDomains.includes(t.table_name))
|
|
}
|
|
|
|
ctx.info(` Found ${filteredTables.length} tables to process`, {
|
|
domains: filteredTables.map((t) => t.table_name)
|
|
})
|
|
|
|
// Process each table
|
|
for (const table of filteredTables) {
|
|
const tableName = table.table_name
|
|
ctx.info(` Processing table: ${tableName}`)
|
|
|
|
const progress = await pgClient`
|
|
SELECT field
|
|
FROM temp_data.account_personid_mapping_v2_progress
|
|
WHERE workspace_id = ${wsUuid} AND domain = ${tableName}
|
|
`
|
|
|
|
const createdDone = !force && progress.some((p) => p.field === 'createdBy')
|
|
const modifiedDone = !force && progress.some((p) => p.field === 'modifiedBy')
|
|
|
|
// Get counts for logging
|
|
const [createdByCount] = !createdDone
|
|
? await pgClient`
|
|
SELECT COUNT(*)
|
|
FROM ${pgClient(tableName)} t
|
|
JOIN temp_data.account_personid_mapping_v2 m ON t."workspaceId" = m.workspace_id AND t."createdBy" = m.old_account_id
|
|
WHERE t."workspaceId" = ${wsUuid}
|
|
`
|
|
: [{ count: 0 }]
|
|
|
|
const [modifiedByCount] = !modifiedDone
|
|
? await pgClient`
|
|
SELECT COUNT(*)
|
|
FROM ${pgClient(tableName)} t
|
|
JOIN temp_data.account_personid_mapping_v2 m ON t."workspaceId" = m.workspace_id AND t."modifiedBy" = m.old_account_id
|
|
WHERE t."workspaceId" = ${wsUuid}
|
|
`
|
|
: [{ count: 0 }]
|
|
|
|
ctx.info(
|
|
` Table ${tableName}: ${createdByCount.count} createdBy and ${modifiedByCount.count} modifiedBy records need updating`
|
|
)
|
|
|
|
if (createdByCount.count > 0) {
|
|
ctx.info(` Updating createdBy for ${tableName}...`)
|
|
const startTime = Date.now()
|
|
|
|
if (batchSize == null || batchSize > createdByCount.count) {
|
|
ctx.info(` Processing the whole table ${tableName}...`)
|
|
await pgClient`
|
|
UPDATE ${pgClient(tableName)}
|
|
SET "createdBy" = m.new_person_id::text
|
|
FROM temp_data.account_personid_mapping_v2 m
|
|
WHERE ${pgClient(tableName)}."workspaceId" = ${wsUuid} AND ${pgClient(tableName)}."workspaceId" = m.workspace_id AND ${pgClient(tableName)}."createdBy" = m.old_account_id
|
|
`
|
|
progressMade = true
|
|
} else {
|
|
ctx.info(` Processing the table ${tableName} in batches of ${batchSize}...`)
|
|
let processed = 0
|
|
while (true) {
|
|
const res = await pgClient`
|
|
UPDATE ${pgClient(tableName)}
|
|
SET "createdBy" = m.new_person_id::text
|
|
FROM temp_data.account_personid_mapping_v2 m
|
|
WHERE ${pgClient(tableName)}."workspaceId" = ${wsUuid} AND ${pgClient(tableName)}."workspaceId" = m.workspace_id AND ${pgClient(tableName)}."createdBy" = m.old_account_id
|
|
LIMIT ${batchSize}
|
|
`
|
|
progressMade = true
|
|
if (res.count === 0) {
|
|
break
|
|
}
|
|
processed += res.count
|
|
const duration = (Date.now() - startTime) / 1000
|
|
const rate = Math.round(processed / duration)
|
|
ctx.info(
|
|
` Processing createdBy for ${tableName}: ${processed} rows in ${duration}s (${rate} rows/sec)`
|
|
)
|
|
}
|
|
}
|
|
|
|
await pgClient`INSERT INTO temp_data.account_personid_mapping_v2_progress (workspace_id, domain, field) VALUES (${wsUuid}, ${tableName}, 'createdBy') ON CONFLICT DO NOTHING`
|
|
|
|
const duration = (Date.now() - startTime) / 1000
|
|
const rate = Math.round(createdByCount.count / duration)
|
|
ctx.info(
|
|
` Updated createdBy for ${tableName}: ${createdByCount.count} rows in ${duration}s (${rate} rows/sec)`
|
|
)
|
|
} else {
|
|
if (createdDone) {
|
|
ctx.info(' Skipping createdBy for table. Already done', { tableName })
|
|
} else {
|
|
await pgClient`INSERT INTO temp_data.account_personid_mapping_v2_progress (workspace_id, domain, field) VALUES (${wsUuid}, ${tableName}, 'createdBy') ON CONFLICT DO NOTHING`
|
|
}
|
|
}
|
|
|
|
if (modifiedByCount.count > 0) {
|
|
ctx.info(` Updating modifiedBy for ${tableName}...`)
|
|
const startTime = Date.now()
|
|
|
|
if (batchSize == null || batchSize > modifiedByCount.count) {
|
|
ctx.info(` Processing the whole table ${tableName}...`)
|
|
await pgClient`
|
|
UPDATE ${pgClient(tableName)}
|
|
SET "modifiedBy" = m.new_person_id::text
|
|
FROM temp_data.account_personid_mapping_v2 m
|
|
WHERE ${pgClient(tableName)}."workspaceId" = ${wsUuid} AND ${pgClient(tableName)}."workspaceId" = m.workspace_id AND ${pgClient(tableName)}."modifiedBy" = m.old_account_id
|
|
`
|
|
progressMade = true
|
|
} else {
|
|
ctx.info(` Processing the table ${tableName} in batches of ${batchSize}...`)
|
|
let processed = 0
|
|
while (true) {
|
|
const res = await pgClient`
|
|
UPDATE ${pgClient(tableName)}
|
|
SET "modifiedBy" = m.new_person_id::text
|
|
FROM temp_data.account_personid_mapping_v2 m
|
|
WHERE ${pgClient(tableName)}."workspaceId" = ${wsUuid} AND ${pgClient(tableName)}."workspaceId" = m.workspace_id AND ${pgClient(tableName)}."modifiedBy" = m.old_account_id
|
|
LIMIT ${batchSize}
|
|
`
|
|
progressMade = true
|
|
if (res.count === 0) {
|
|
break
|
|
}
|
|
processed += res.count
|
|
const duration = (Date.now() - startTime) / 1000
|
|
const rate = Math.round(processed / duration)
|
|
ctx.info(
|
|
` Processing modifiedBy for ${tableName}: ${processed} rows in ${duration}s (${rate} rows/sec)`
|
|
)
|
|
}
|
|
}
|
|
|
|
await pgClient`INSERT INTO temp_data.account_personid_mapping_v2_progress (workspace_id, domain, field) VALUES (${wsUuid}, ${tableName}, 'modifiedBy') ON CONFLICT DO NOTHING`
|
|
|
|
const duration = (Date.now() - startTime) / 1000
|
|
const rate = Math.round(modifiedByCount.count / duration)
|
|
ctx.info(
|
|
` Updated modifiedBy for ${tableName}: ${modifiedByCount.count} rows in ${duration}s (${rate} rows/sec)`
|
|
)
|
|
} else {
|
|
if (modifiedDone) {
|
|
ctx.info(' Skipping modifiedBy for table. Already done', { tableName })
|
|
} else {
|
|
await pgClient`INSERT INTO temp_data.account_personid_mapping_v2_progress (workspace_id, domain, field) VALUES (${wsUuid}, ${tableName}, 'modifiedBy') ON CONFLICT DO NOTHING`
|
|
}
|
|
}
|
|
}
|
|
|
|
done = true
|
|
ctx.info('Migration of created/modified completed successfully')
|
|
} catch (err: any) {
|
|
if (err.code === '40001' || err.code === '55P03') {
|
|
// Retry transaction
|
|
if (retriesCount === maxRetries) {
|
|
ctx.error('Failed to migrate created/modified by. Max retries reached', { err })
|
|
} else {
|
|
retrying = true
|
|
continue
|
|
}
|
|
}
|
|
|
|
if (err.code === 'CONNECTION_CLOSED') {
|
|
// Reconnect
|
|
ctx.info(' Connection closed...')
|
|
if (connectsCount === maxReconnects) {
|
|
ctx.error('Failed to migrate created/modified by. Max reconnects reached', { err })
|
|
} else {
|
|
reconnecting = true
|
|
continue
|
|
}
|
|
}
|
|
|
|
throw err
|
|
} finally {
|
|
pg?.close()
|
|
}
|
|
}
|
|
|
|
if (!done) {
|
|
ctx.error('Failed to migrate created/modified by')
|
|
}
|
|
}
|
|
|
|
async function fillAccountSocialKeyMapping (ctx: MeasureMetricsContext, pgClient: postgres.Sql): Promise<void> {
|
|
ctx.info('Creating account to social key mapping table...')
|
|
// Create schema
|
|
await pgClient`CREATE SCHEMA IF NOT EXISTS temp_data`
|
|
|
|
// Create mapping table
|
|
await pgClient`
|
|
CREATE TABLE IF NOT EXISTS temp_data.account_socialkey_mapping (
|
|
workspace_id text,
|
|
old_account_id text,
|
|
new_social_key text,
|
|
person_ref text,
|
|
person_name text,
|
|
INDEX idx_account_mapping_old_id (workspace_id, old_account_id)
|
|
)
|
|
`
|
|
|
|
const [res] = await pgClient`SELECT COUNT(*) FROM temp_data.account_socialkey_mapping`
|
|
|
|
if (res.count === '0') {
|
|
// Populate mapping table
|
|
await pgClient`
|
|
INSERT INTO temp_data.account_socialkey_mapping
|
|
WITH accounts AS (
|
|
SELECT
|
|
tx."workspaceId",
|
|
tx."objectId",
|
|
COALESCE(
|
|
-- Get the latest email from updates
|
|
(
|
|
SELECT tx2.data->'operations'->>'email'
|
|
FROM model_tx tx2
|
|
WHERE tx2."objectId" = tx."objectId"
|
|
AND tx2."workspaceId" = tx."workspaceId"
|
|
AND tx2.data->>'objectClass' = 'contact:class:PersonAccount'
|
|
AND tx2.data->'operations'->>'email' IS NOT NULL
|
|
ORDER BY tx2."createdOn" DESC
|
|
LIMIT 1
|
|
),
|
|
-- If no updates with email, get from create transaction
|
|
tx.data->'attributes'->>'email'
|
|
) as latest_email,
|
|
COALESCE(
|
|
-- Get the latest person from updates
|
|
(
|
|
SELECT (tx2.data->'operations'->>'person')::text
|
|
FROM model_tx tx2
|
|
WHERE tx2."objectId" = tx."objectId"
|
|
AND tx2."workspaceId" = tx."workspaceId"
|
|
AND tx2.data->>'objectClass' = 'contact:class:PersonAccount'
|
|
AND tx2.data->'operations'->>'person' IS NOT NULL
|
|
ORDER BY tx2."createdOn" DESC
|
|
LIMIT 1
|
|
),
|
|
-- If no updates, get from create transaction
|
|
(tx.data->'attributes'->>'person')::text
|
|
) as person_ref
|
|
FROM model_tx tx
|
|
WHERE tx."_class" = 'core:class:TxCreateDoc'
|
|
AND tx.data->>'objectClass' = 'contact:class:PersonAccount'
|
|
AND tx."objectId" NOT IN ('core:account:System', 'core:account:ConfigUser')
|
|
)
|
|
SELECT
|
|
a."workspaceId" as workspace_id,
|
|
a."objectId" as old_account_id,
|
|
CASE
|
|
WHEN a.latest_email LIKE 'github:%' THEN lower(a.latest_email)
|
|
WHEN a.latest_email LIKE 'openid:%' THEN 'oidc:' || lower(substring(a.latest_email from 8))
|
|
ELSE 'email:' || lower(a.latest_email)
|
|
END as new_social_key,
|
|
a.person_ref,
|
|
c.data->>'name' as person_name
|
|
FROM accounts as a
|
|
LEFT JOIN public.contact c ON c."_id" = a.person_ref AND c."workspaceId" = a."workspaceId"
|
|
WHERE a.latest_email IS NOT NULL
|
|
AND a.latest_email != ''
|
|
`
|
|
}
|
|
}
|
|
|
|
export async function migrateMergedAccounts (
|
|
ctx: MeasureMetricsContext,
|
|
dbUrl: string,
|
|
accountDb: AccountDB
|
|
): Promise<void> {
|
|
ctx.info('Migrating merged person accounts... ', {})
|
|
|
|
if (!dbUrl.startsWith('postgresql')) {
|
|
throw new Error('Only CockroachDB is supported')
|
|
}
|
|
|
|
const pg = getDBClient(dbUrl)
|
|
const pgClient = await pg.getClient()
|
|
const token = getToolToken()
|
|
|
|
try {
|
|
await fillAccountSocialKeyMapping(ctx, pgClient)
|
|
|
|
const personsAccounts = await pgClient`
|
|
SELECT workspace_id, person_ref, array_agg(new_social_key) as social_keys
|
|
FROM temp_data.account_socialkey_mapping
|
|
WHERE new_social_key != 'email:huly.ai.bot@hc.engineering'
|
|
GROUP BY workspace_id, person_ref
|
|
HAVING count(*) > 1
|
|
`
|
|
|
|
ctx.info('Processing persons with merged accounts ', { count: personsAccounts.length })
|
|
let processed = 0
|
|
let errors = 0
|
|
|
|
for (const personAccounts of personsAccounts) {
|
|
try {
|
|
const socialKeys = personAccounts.social_keys
|
|
|
|
// Every social id in the old account might either be already in the new account or not in the accounts at all
|
|
// So we want to
|
|
// 1. Take the first social id with the existing account
|
|
// 2. Merge all other accounts into the first one
|
|
// 3. Create social ids for the first account which haven't had their own accounts
|
|
const toAdd = new Set<SocialKey>()
|
|
const toMergePersons = new Set<PersonUuid>()
|
|
const toMergeAccounts = new Set<AccountUuid>()
|
|
for (const socialKey of socialKeys) {
|
|
const socialIdKey = parseSocialIdString(socialKey)
|
|
const socialId = await findFullSocialIdBySocialKey(ctx, accountDb, null, token, { socialKey })
|
|
const personUuid = socialId?.personUuid
|
|
const accountUuid = (await findPersonBySocialKey(ctx, accountDb, null, token, {
|
|
socialString: socialKey,
|
|
requireAccount: true
|
|
})) as AccountUuid
|
|
|
|
if (personUuid == null) {
|
|
toAdd.add(socialIdKey)
|
|
// Means not attached to any account yet, simply add the social id to the primary account
|
|
} else if (accountUuid == null) {
|
|
toMergePersons.add(personUuid)
|
|
} else {
|
|
// This is the case when the social id is already attached to an account. Merge the accounts.
|
|
toMergeAccounts.add(accountUuid)
|
|
}
|
|
}
|
|
|
|
if (toMergeAccounts.size === 0) {
|
|
// No existing accounts for the person's social ids. Normally this should never be the case.
|
|
ctx.info('No existing accounts for person', personAccounts)
|
|
continue
|
|
}
|
|
|
|
const toMergeAccountsArray = Array.from(toMergeAccounts)
|
|
const primaryAccount = toMergeAccountsArray[0]
|
|
|
|
for (let i = 1; i < toMergeAccountsArray.length; i++) {
|
|
const accountToMerge = toMergeAccountsArray[i]
|
|
await mergeSpecifiedAccounts(ctx, accountDb, null, token, {
|
|
primaryAccount,
|
|
secondaryAccount: accountToMerge
|
|
})
|
|
}
|
|
|
|
const toMergePersonsArray = Array.from(toMergePersons)
|
|
for (const personToMerge of toMergePersonsArray) {
|
|
await mergeSpecifiedPersons(ctx, accountDb, null, token, {
|
|
primaryPerson: primaryAccount,
|
|
secondaryPerson: personToMerge
|
|
})
|
|
}
|
|
|
|
for (const addTarget of Array.from(toAdd)) {
|
|
await addSocialIdToPerson(ctx, accountDb, null, token, {
|
|
person: primaryAccount,
|
|
...addTarget,
|
|
confirmed: false
|
|
})
|
|
}
|
|
|
|
processed++
|
|
if (processed % 10 === 0) {
|
|
ctx.info(`Processed ${processed} of ${personsAccounts.length} persons`)
|
|
}
|
|
} catch (err: any) {
|
|
errors++
|
|
ctx.error('Failed to merge accounts for person', { mergedGroup: personAccounts, err })
|
|
}
|
|
}
|
|
|
|
ctx.info('Finished processing persons with merged accounts', { processed, of: personsAccounts.length, errors })
|
|
} catch (err: any) {
|
|
ctx.error('Failed to migrate merged accounts', { err })
|
|
} finally {
|
|
pg.close()
|
|
}
|
|
}
|
|
|
|
export async function filterMergedAccountsInMembers (
|
|
ctx: MeasureMetricsContext,
|
|
dbUrl: string,
|
|
accountDb: AccountDB
|
|
): Promise<void> {
|
|
ctx.info('Filtering merged accounts in members... ', {})
|
|
|
|
if (!dbUrl.startsWith('postgresql')) {
|
|
throw new Error('Only CockroachDB is supported')
|
|
}
|
|
|
|
const pg = getDBClient(dbUrl)
|
|
const pgClient = await pg.getClient()
|
|
|
|
try {
|
|
const mergedPersons = await accountDb.person.find({ migratedTo: { $ne: null } })
|
|
|
|
if (mergedPersons.length === 0) {
|
|
ctx.info('No merged persons to migrate')
|
|
return
|
|
}
|
|
|
|
ctx.info('Merged persons found', { count: mergedPersons.length })
|
|
|
|
const migrationMap = new Map<PersonUuid, PersonUuid>()
|
|
for (const person of mergedPersons) {
|
|
if (person.migratedTo == null) {
|
|
continue
|
|
}
|
|
|
|
migrationMap.set(person.uuid, person.migratedTo)
|
|
}
|
|
|
|
const spacesToUpdate = await pgClient`
|
|
SELECT "workspaceId", _id, members FROM ${pgClient(DOMAIN_SPACE)} WHERE members && ${pgClient.array(Array.from(migrationMap.keys()))}
|
|
`
|
|
|
|
ctx.info('Spaces to update', { count: spacesToUpdate.length })
|
|
|
|
let processed = 0
|
|
let errors = 0
|
|
for (const space of spacesToUpdate) {
|
|
try {
|
|
const newMembers = new Set<PersonUuid>(space.members.map((it: PersonUuid) => migrationMap.get(it) ?? it))
|
|
|
|
await pgClient`
|
|
UPDATE ${pgClient(DOMAIN_SPACE)} SET members = ${pgClient.array(Array.from(newMembers))}
|
|
WHERE "workspaceId" = ${space.workspaceId}
|
|
AND "_id" = ${space._id}
|
|
`
|
|
processed++
|
|
} catch (err: any) {
|
|
errors++
|
|
ctx.error('Failed to update space members', { space, err })
|
|
}
|
|
}
|
|
|
|
ctx.info('Finished updating spaces', { processed, of: spacesToUpdate.length, errors })
|
|
} finally {
|
|
pg.close()
|
|
}
|
|
}
|
|
|
|
export async function ensureGlobalPersonsForLocalAccounts (
|
|
ctx: MeasureMetricsContext,
|
|
dbUrl: string,
|
|
accountDb: AccountDB
|
|
): Promise<void> {
|
|
ctx.info('Ensuring global persons for local accounts... ', {})
|
|
|
|
if (!dbUrl.startsWith('postgresql')) {
|
|
throw new Error('Only CockroachDB is supported')
|
|
}
|
|
|
|
const pg = getDBClient(dbUrl)
|
|
const pgClient = await pg.getClient()
|
|
const token = getToolToken()
|
|
|
|
try {
|
|
await fillAccountSocialKeyMapping(ctx, pgClient)
|
|
|
|
let count = 0
|
|
let failed = 0
|
|
const accountToSocialKey = await pgClient`SELECT * FROM temp_data.account_socialkey_mapping`
|
|
for (const row of accountToSocialKey) {
|
|
const newSocialKey = row.new_social_key
|
|
const personName = row.person_name ?? ''
|
|
|
|
const keyParts = newSocialKey.split(':')
|
|
if (keyParts.length !== 2) {
|
|
ctx.error('Invalid social key', row)
|
|
continue
|
|
}
|
|
|
|
const keyType = keyParts[0]
|
|
const keyValue = keyParts[1]
|
|
|
|
if (!Object.values(SocialIdType).includes(keyType)) {
|
|
ctx.error('Invalid social key type', row)
|
|
continue
|
|
}
|
|
|
|
const firstName = getFirstName(personName)
|
|
const lastName = getLastName(personName)
|
|
const effectiveFirstName = firstName === '' ? keyValue : firstName
|
|
|
|
try {
|
|
await ensurePerson(ctx, accountDb, null, token, {
|
|
socialType: keyType as SocialIdType,
|
|
socialValue: keyValue,
|
|
firstName: effectiveFirstName,
|
|
lastName
|
|
})
|
|
count++
|
|
} catch (err: any) {
|
|
ctx.error('Failed to ensure person', {
|
|
socialType: keyType as SocialIdType,
|
|
socialValue: keyValue,
|
|
firstName: effectiveFirstName,
|
|
lastName
|
|
})
|
|
failed++
|
|
}
|
|
}
|
|
|
|
ctx.info(`Successfully ensured ${count} people with failed count ${failed}`)
|
|
} finally {
|
|
pg.close()
|
|
}
|
|
}
|
|
|
|
export async function migrateTrustedV6Accounts (
|
|
ctx: MeasureMetricsContext,
|
|
accountDB: AccountDB,
|
|
mongoDb: v6MongoAccountDB,
|
|
dryRun: boolean,
|
|
skipWorkspaces: Set<string>,
|
|
conflictSuffix?: string
|
|
): Promise<void> {
|
|
// Mapping between <ObjectId, UUID>
|
|
const accountsIdToUuid: Record<string, AccountUuid> = {}
|
|
// Mapping between <email, UUID>
|
|
const accountsEmailToUuid: Record<string, AccountUuid> = {}
|
|
// Mapping between <OldId, UUID>
|
|
const workspacesIdToUuid: Record<WorkspaceDataId, WorkspaceUuid> = {}
|
|
|
|
console.log('Migrating accounts database...')
|
|
let accountsProcessed = 0
|
|
const accountsCursor = mongoDb.account.findCursor({})
|
|
try {
|
|
while (await accountsCursor.hasNext()) {
|
|
const account = await accountsCursor.next()
|
|
if (account == null) {
|
|
break
|
|
}
|
|
|
|
try {
|
|
const accountUuid = await migrateAccount(account, accountDB, dryRun)
|
|
if (accountUuid == null) {
|
|
console.log('Account not migrated', account)
|
|
continue
|
|
}
|
|
accountsIdToUuid[account._id.toString()] = accountUuid
|
|
accountsEmailToUuid[account.email] = accountUuid
|
|
|
|
accountsProcessed++
|
|
if (accountsProcessed % 100 === 0) {
|
|
console.log('Processed accounts:', accountsProcessed)
|
|
}
|
|
} catch (err: any) {
|
|
console.log('Failed to migrate account', account._id, account.email, err)
|
|
}
|
|
}
|
|
} catch (err: any) {
|
|
console.log('Failed to migrate accounts', err)
|
|
} finally {
|
|
await accountsCursor.close()
|
|
}
|
|
|
|
console.log('Total accounts processed:', accountsProcessed)
|
|
|
|
let processedWorkspaces = 0
|
|
const workspacesCursor = mongoDb.workspace.findCursor({})
|
|
try {
|
|
while (await workspacesCursor.hasNext()) {
|
|
const workspace = await workspacesCursor.next()
|
|
if (workspace == null) {
|
|
break
|
|
}
|
|
|
|
if (
|
|
skipWorkspaces.has(workspace.workspace) ||
|
|
(workspace.workspaceUrl != null && skipWorkspaces.has(workspace.workspaceUrl))
|
|
) {
|
|
console.log('Skipping workspace', workspace.workspace, workspace.workspaceUrl)
|
|
continue
|
|
}
|
|
|
|
try {
|
|
const [workspaceUuid] =
|
|
(await migrateWorkspace(
|
|
workspace,
|
|
accountDB,
|
|
accountsIdToUuid,
|
|
accountsEmailToUuid,
|
|
dryRun,
|
|
'manual-creation',
|
|
conflictSuffix
|
|
)) ?? []
|
|
|
|
if (workspaceUuid !== undefined) {
|
|
workspacesIdToUuid[workspace.workspace] = workspaceUuid
|
|
}
|
|
processedWorkspaces++
|
|
if (processedWorkspaces % 100 === 0) {
|
|
console.log('Processed workspaces:', processedWorkspaces)
|
|
}
|
|
} catch (err: any) {
|
|
console.log('Failed to migrate workspace', workspace.workspaceUrl, workspace.workspace, err)
|
|
}
|
|
}
|
|
} catch (err: any) {
|
|
console.log('Failed to migrate workspaces', err)
|
|
} finally {
|
|
await workspacesCursor.close()
|
|
}
|
|
|
|
console.log('Total workspaces processed:', processedWorkspaces)
|
|
console.log('Total workspaces created/ensured:', Object.values(workspacesIdToUuid).length)
|
|
|
|
let invitesProcessed = 0
|
|
const invitesCursor = mongoDb.invite.findCursor({})
|
|
try {
|
|
while (await invitesCursor.hasNext()) {
|
|
const invite = await invitesCursor.next()
|
|
if (invite == null) {
|
|
break
|
|
}
|
|
|
|
try {
|
|
const workspaceUuid = workspacesIdToUuid[invite.workspace.name]
|
|
if (workspaceUuid === undefined) {
|
|
console.log('No workspace with id', invite.workspace.name, 'found for invite', invite._id)
|
|
continue
|
|
}
|
|
|
|
const existing = await accountDB.invite.findOne({ migratedFrom: invite._id.toString() })
|
|
if (existing != null) {
|
|
continue
|
|
}
|
|
|
|
const inviteRecord = {
|
|
migratedFrom: invite._id.toString(),
|
|
workspaceUuid,
|
|
expiresOn: invite.exp,
|
|
emailPattern: invite.emailMask,
|
|
remainingUses: invite.limit,
|
|
role: invite.role ?? AccountRole.User
|
|
}
|
|
|
|
if (!dryRun) {
|
|
await accountDB.invite.insertOne(inviteRecord)
|
|
} else {
|
|
console.log('Creating invite record', inviteRecord)
|
|
}
|
|
|
|
invitesProcessed++
|
|
if (invitesProcessed % 100 === 0) {
|
|
console.log('Processed invites:', invitesProcessed)
|
|
}
|
|
} catch (err: any) {
|
|
console.log('Failed to migrate invite', invite._id, err)
|
|
}
|
|
}
|
|
} catch (err: any) {
|
|
console.log('Failed to migrate invites', err)
|
|
} finally {
|
|
await invitesCursor.close()
|
|
}
|
|
|
|
console.log('Total invites processed:', invitesProcessed)
|
|
}
|
|
|
|
async function migrateAccount (
|
|
account: OldAccount,
|
|
accountDB: AccountDB,
|
|
dryRun = true
|
|
): Promise<AccountUuid | undefined> {
|
|
let primaryKey: SocialKey
|
|
let secondaryKey: SocialKey | undefined
|
|
|
|
if (account.githubId != null) {
|
|
if (account.githubUser == null) {
|
|
console.log('No github user found for github id', account.githubId)
|
|
return
|
|
}
|
|
|
|
primaryKey = {
|
|
type: SocialIdType.GITHUB,
|
|
value: account.githubUser
|
|
}
|
|
secondaryKey = !account.email.startsWith('github:')
|
|
? {
|
|
type: SocialIdType.EMAIL,
|
|
value: account.email
|
|
}
|
|
: undefined
|
|
} else if (account.openId != null) {
|
|
primaryKey = {
|
|
type: SocialIdType.OIDC,
|
|
value: account.openId
|
|
}
|
|
secondaryKey = !account.email.startsWith('openid:')
|
|
? {
|
|
type: SocialIdType.EMAIL,
|
|
value: account.email
|
|
}
|
|
: undefined
|
|
} else {
|
|
primaryKey = {
|
|
type: SocialIdType.EMAIL,
|
|
value: account.email
|
|
}
|
|
}
|
|
|
|
let personUuid: PersonUuid
|
|
const verified = account.confirmed === true ? { verifiedOn: Date.now() } : {}
|
|
|
|
const existing = await accountDB.socialId.findOne(primaryKey)
|
|
if (existing == null) {
|
|
// Create new global person
|
|
const personRecord = {
|
|
firstName: account.first,
|
|
lastName: account.last
|
|
}
|
|
|
|
if (!dryRun) {
|
|
personUuid = await accountDB.person.insertOne(personRecord)
|
|
} else {
|
|
console.log('Creating person record', personRecord)
|
|
personUuid = generateUuid() as PersonUuid
|
|
}
|
|
|
|
const socialIdRecord = {
|
|
...primaryKey,
|
|
personUuid,
|
|
...verified
|
|
}
|
|
|
|
if (!dryRun) {
|
|
await accountDB.socialId.insertOne(socialIdRecord)
|
|
} else {
|
|
console.log('Creating social id record', socialIdRecord)
|
|
}
|
|
|
|
if (!dryRun) {
|
|
await createAccount(accountDB, personUuid, account.confirmed, false, account.createdOn)
|
|
} else {
|
|
console.log('Creating account record', { personUuid, confirmed: account.confirmed })
|
|
}
|
|
|
|
if (account.hash != null && account.salt != null) {
|
|
if (!dryRun) {
|
|
await accountDB.setPassword(personUuid as AccountUuid, account.hash, account.salt)
|
|
} else {
|
|
console.log('Updating account password', { personUuid })
|
|
}
|
|
}
|
|
} else {
|
|
personUuid = existing.personUuid
|
|
|
|
// if there's no existing account, create a new one
|
|
const existingAcc = await accountDB.account.findOne({ uuid: personUuid as AccountUuid })
|
|
if (existingAcc == null) {
|
|
if (!dryRun) {
|
|
await createAccount(accountDB, personUuid, account.confirmed, false, account.createdOn)
|
|
} else {
|
|
console.log('Creating account record', { personUuid, confirmed: account.confirmed })
|
|
}
|
|
|
|
if (account.hash != null && account.salt != null) {
|
|
if (!dryRun) {
|
|
await accountDB.setPassword(personUuid as AccountUuid, account.hash, account.salt)
|
|
} else {
|
|
console.log('Updating account password', { personUuid })
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
if (secondaryKey != null) {
|
|
const existingSecondary = await accountDB.socialId.findOne(secondaryKey)
|
|
if (existingSecondary == null) {
|
|
if (!dryRun) {
|
|
await accountDB.socialId.insertOne({
|
|
...secondaryKey,
|
|
personUuid,
|
|
...verified
|
|
})
|
|
} else {
|
|
console.log('Creating secondary social id', { personUuid, confirmed: account.confirmed, secondaryKey })
|
|
}
|
|
}
|
|
}
|
|
|
|
return personUuid as AccountUuid
|
|
}
|
|
|
|
async function migrateWorkspace (
|
|
workspace: OldWorkspace,
|
|
accountDB: AccountDB,
|
|
accountsIdToUuid: Record<string, AccountUuid>,
|
|
accountsEmailToUuid: Record<string, AccountUuid>,
|
|
dryRun = true,
|
|
forcedMode?: WorkspaceMode,
|
|
conflictSuffix?: string,
|
|
throwExisting?: boolean,
|
|
region?: string,
|
|
branding?: string
|
|
): Promise<[WorkspaceUuid, string] | undefined> {
|
|
if (workspace.workspaceUrl == null) {
|
|
console.log('No workspace url, skipping', workspace.workspace)
|
|
return
|
|
}
|
|
|
|
const createdBy = workspace.createdBy !== undefined ? accountsEmailToUuid[workspace.createdBy] : undefined
|
|
if (createdBy === undefined) {
|
|
console.log('No account found for workspace', workspace.workspace, 'created by', workspace.createdBy)
|
|
}
|
|
|
|
let existingByUrl = await accountDB.workspace.findOne({ url: workspace.workspaceUrl })
|
|
const existingByUuid = await accountDB.workspace.findOne({ uuid: workspace.uuid })
|
|
|
|
let workspaceUuid: WorkspaceUuid
|
|
let url = workspace.workspaceUrl
|
|
|
|
if (existingByUuid == null) {
|
|
if (existingByUrl != null && (conflictSuffix ?? '') !== '') {
|
|
url = `${url}-${conflictSuffix}`
|
|
existingByUrl = await accountDB.workspace.findOne({ url })
|
|
}
|
|
if (existingByUrl != null) {
|
|
console.log('Conflicting workspace url', url)
|
|
// generate new url
|
|
url = `${url}-${generateId('-')}`
|
|
console.log('Generating new url', url)
|
|
}
|
|
|
|
const workspaceRecord = {
|
|
uuid: workspace.uuid,
|
|
name: workspace.workspaceName,
|
|
url,
|
|
dataId: workspace.workspace,
|
|
branding: branding ?? workspace.branding,
|
|
region: region ?? workspace.region,
|
|
createdBy,
|
|
billingAccount: createdBy,
|
|
createdOn: workspace.createdOn ?? Date.now()
|
|
}
|
|
|
|
if (!dryRun) {
|
|
workspaceUuid = await accountDB.workspace.insertOne(workspaceRecord)
|
|
} else {
|
|
console.log('Creating workspace record', workspaceRecord)
|
|
workspaceUuid = generateUuid() as WorkspaceUuid
|
|
}
|
|
} else {
|
|
if (throwExisting === true) {
|
|
throw new Error(`Workspace with the same uuid ${workspace.uuid} already exists`)
|
|
}
|
|
|
|
workspaceUuid = existingByUuid.uuid
|
|
}
|
|
|
|
const existingStatus = await accountDB.workspaceStatus.findOne({ workspaceUuid })
|
|
|
|
if (existingStatus == null) {
|
|
const statusRecord = {
|
|
workspaceUuid,
|
|
mode: forcedMode ?? workspace.mode,
|
|
processingProgress: workspace.progress !== undefined ? Math.floor(workspace.progress) : undefined,
|
|
versionMajor: workspace.version?.major,
|
|
versionMinor: workspace.version?.minor,
|
|
versionPatch: workspace.version?.patch,
|
|
lastProcessingTime: workspace.lastProcessingTime,
|
|
lastVisit: workspace.lastVisit,
|
|
isDisabled: workspace.disabled,
|
|
processingAttempts: workspace.attempts,
|
|
processingMessage: workspace.message,
|
|
backupInfo: workspace.backupInfo
|
|
}
|
|
|
|
if (!dryRun) {
|
|
await accountDB.workspaceStatus.insertOne(statusRecord)
|
|
} else {
|
|
console.log('Creating workspace status record', statusRecord)
|
|
}
|
|
}
|
|
|
|
const uniqueAccounts = Array.from(new Set((workspace.accounts ?? []).map((it) => it.toString())))
|
|
const existingMembers = new Set((await accountDB.getWorkspaceMembers(workspaceUuid)).map((mi) => mi.person))
|
|
for (const member of uniqueAccounts) {
|
|
const accountUuid = accountsIdToUuid[member]
|
|
|
|
if (accountUuid === undefined) {
|
|
console.log('No account found for workspace', workspace.workspace, 'member', member)
|
|
continue
|
|
}
|
|
|
|
if (existingMembers.has(accountUuid)) {
|
|
continue
|
|
}
|
|
|
|
if (!dryRun) {
|
|
// Actual roles are being set in workspace migration
|
|
await accountDB.assignWorkspace(accountUuid, workspaceUuid, AccountRole.Guest)
|
|
} else {
|
|
console.log('Assigning account', member, accountUuid, 'to workspace', workspaceUuid)
|
|
}
|
|
}
|
|
|
|
return [workspaceUuid, url]
|
|
}
|
|
|
|
export async function restoreFromv6All (
|
|
ctx: MeasureMetricsContext,
|
|
accountDB: AccountDB,
|
|
dirName: string,
|
|
txes: Tx[],
|
|
dbUrl: string
|
|
): Promise<void> {
|
|
ctx.info('Restoring from v6 dump...')
|
|
const v6AccountsFile = 'account.accounts.json'
|
|
const v6WorkspacesFile = 'account.workspaces.json'
|
|
const v6InvitesFile = 'account.invites.json'
|
|
|
|
const storage = await createFileBackupStorage(dirName)
|
|
|
|
if (!(await storage.exists(v6AccountsFile))) {
|
|
ctx.error('file not pressent', { file: v6AccountsFile })
|
|
throw new Error(`${v6AccountsFile} should present to restore`)
|
|
}
|
|
if (!(await storage.exists(v6WorkspacesFile))) {
|
|
ctx.error('file not pressent', { file: v6WorkspacesFile })
|
|
throw new Error(`${v6WorkspacesFile} should present to restore`)
|
|
}
|
|
if (!(await storage.exists(v6InvitesFile))) {
|
|
ctx.error('file not pressent', { file: v6InvitesFile })
|
|
throw new Error(`${v6InvitesFile} should present to restore`)
|
|
}
|
|
|
|
try {
|
|
const v6AccountsRaw = JSON.parse((await storage.loadFile(v6AccountsFile)).toString())
|
|
const v6Workspaces = JSON.parse((await storage.loadFile(v6WorkspacesFile)).toString()) as OldWorkspace[]
|
|
const v6Invites = JSON.parse((await storage.loadFile(v6InvitesFile)).toString())
|
|
|
|
const v6Accounts: OldAccount[] = []
|
|
for (const rawAccount of v6AccountsRaw) {
|
|
const hashTypedArray = rawAccount.hash != null ? new Uint8Array(rawAccount.hash.data) : null
|
|
const saltTypedArray = new Uint8Array(rawAccount.salt.data)
|
|
|
|
v6Accounts.push({
|
|
...rawAccount,
|
|
hash: hashTypedArray != null ? Buffer.from(hashTypedArray.buffer) : null,
|
|
salt: Buffer.from(saltTypedArray.buffer)
|
|
})
|
|
}
|
|
|
|
// Generate UUIDs for workspaces where missing
|
|
for (const workspace of v6Workspaces) {
|
|
if (workspace.uuid == null) {
|
|
workspace.uuid = generateUuid() as WorkspaceUuid
|
|
}
|
|
}
|
|
|
|
// Mapping between <ObjectId, UUID>
|
|
const accountsIdToUuid: Record<string, AccountUuid> = {}
|
|
// Mapping between <email, UUID>
|
|
const accountsEmailToUuid: Record<string, AccountUuid> = {}
|
|
// Mapping between <OldId, UUID>
|
|
const workspacesIdToUuid: Record<WorkspaceDataId, WorkspaceUuid> = {}
|
|
|
|
ctx.info('Restoring accounts database...')
|
|
|
|
let accountsProcessed = 0
|
|
for (const account of v6Accounts) {
|
|
try {
|
|
const accountUuid = await migrateAccount(account, accountDB, false)
|
|
if (accountUuid == null) {
|
|
ctx.warn('Account not restored', account)
|
|
continue
|
|
}
|
|
|
|
accountsIdToUuid[account._id.toString()] = accountUuid
|
|
accountsEmailToUuid[account.email] = accountUuid
|
|
|
|
accountsProcessed++
|
|
if (accountsProcessed % 100 === 0) {
|
|
ctx.info('Processed accounts:', { accountsProcessed })
|
|
}
|
|
} catch (err: any) {
|
|
ctx.error('Failed to restore account', { _id: account._id, email: account.email, err })
|
|
}
|
|
}
|
|
|
|
ctx.info('Total accounts processed:', { accountsProcessed })
|
|
|
|
let processedWorkspaces = 0
|
|
const activeWorkspaces = new Set<WorkspaceUuid>()
|
|
for (const workspace of v6Workspaces) {
|
|
const isActive = workspace.mode === 'active'
|
|
|
|
try {
|
|
// Create active workspaces as archived until they are actually restored
|
|
const [workspaceUuid] =
|
|
(await migrateWorkspace(
|
|
workspace,
|
|
accountDB,
|
|
accountsIdToUuid,
|
|
accountsEmailToUuid,
|
|
false,
|
|
isActive ? 'archived' : undefined
|
|
)) ?? []
|
|
|
|
if (workspaceUuid !== undefined) {
|
|
workspacesIdToUuid[workspace.workspace] = workspaceUuid
|
|
|
|
if (isActive) {
|
|
activeWorkspaces.add(workspaceUuid)
|
|
}
|
|
}
|
|
processedWorkspaces++
|
|
if (processedWorkspaces % 100 === 0) {
|
|
ctx.info('Processed workspaces:', { processedWorkspaces })
|
|
}
|
|
} catch (err: any) {
|
|
ctx.error('Failed to restore workspace', { url: workspace.workspaceUrl, workspace: workspace.workspace, err })
|
|
}
|
|
}
|
|
|
|
ctx.info('Total workspaces processed:', { processedWorkspaces })
|
|
ctx.info('Total workspaces created/ensured:', { count: Object.values(workspacesIdToUuid).length })
|
|
|
|
let invitesProcessed = 0
|
|
for (const invite of v6Invites) {
|
|
try {
|
|
const workspaceUuid = workspacesIdToUuid[invite.workspace.name]
|
|
if (workspaceUuid === undefined) {
|
|
ctx.error(`No workspace with id ${invite.workspace.name} found for invite ${invite._id}`)
|
|
continue
|
|
}
|
|
|
|
const existing = await accountDB.invite.findOne({ migratedFrom: invite._id.toString() })
|
|
if (existing != null) {
|
|
continue
|
|
}
|
|
|
|
const inviteRecord = {
|
|
migratedFrom: invite._id.toString(),
|
|
workspaceUuid,
|
|
expiresOn: invite.exp,
|
|
emailPattern: invite.emailMask,
|
|
remainingUses: invite.limit,
|
|
role: invite.role ?? AccountRole.User
|
|
}
|
|
|
|
await accountDB.invite.insertOne(inviteRecord)
|
|
|
|
invitesProcessed++
|
|
if (invitesProcessed % 100 === 0) {
|
|
ctx.info('Processed invites:', { invitesProcessed })
|
|
}
|
|
} catch (err: any) {
|
|
ctx.error('Failed to restore invite', { _id: invite._id, err })
|
|
}
|
|
}
|
|
|
|
ctx.info('Total invites processed:', { invitesProcessed })
|
|
ctx.info('Successfully restored accounts backup')
|
|
ctx.info('Restoring workspaces...')
|
|
|
|
for (const workspace of v6Workspaces) {
|
|
const dataId = workspace.workspace
|
|
const url = workspace.workspaceUrl
|
|
const uuid = workspacesIdToUuid[dataId]
|
|
|
|
if (url == null) {
|
|
ctx.error('Workspace url not set', { dataId })
|
|
continue
|
|
}
|
|
|
|
if (uuid == null) {
|
|
ctx.error('Workspace uuid not found', { dataId })
|
|
continue
|
|
}
|
|
|
|
const wsIds = {
|
|
uuid,
|
|
dataId,
|
|
url
|
|
}
|
|
|
|
const storage = await createFileBackupStorage(`${dirName}/${dataId}`)
|
|
const storageConfig = storageConfigFromEnv()
|
|
|
|
const queue = getPlatformQueue('tool', workspace.region)
|
|
const wsProducer = queue.getProducer<QueueWorkspaceMessage>(ctx, QueueTopic.Workspace)
|
|
|
|
await wsProducer.send(ctx, uuid, [workspaceEvents.restoring()])
|
|
|
|
const workspaceStorage: StorageAdapter = buildStorageFromConfig(storageConfig)
|
|
|
|
let pipeline: Pipeline | undefined
|
|
try {
|
|
pipeline = await createBackupPipeline(ctx, dbUrl, txes, {
|
|
externalStorage: workspaceStorage,
|
|
usePassedCtx: true
|
|
})(ctx, wsIds, createEmptyBroadcastOps(), null)
|
|
if (pipeline === undefined) {
|
|
ctx.error('failed to restore, pipeline is undefined', { dataId })
|
|
return
|
|
}
|
|
await sendTransactorEvent(uuid, 'force-maintenance')
|
|
|
|
await restore(ctx, pipeline, wsIds, storage, {
|
|
date: -1,
|
|
merge: false,
|
|
parallel: 1,
|
|
recheck: false
|
|
})
|
|
|
|
await sendTransactorEvent(uuid, 'force-close')
|
|
|
|
ctx.info('workspace restored', { dataId })
|
|
await wsProducer.send(ctx, uuid, [workspaceEvents.restored()])
|
|
|
|
if (activeWorkspaces.has(uuid)) {
|
|
// set workspace back to active
|
|
await accountDB.workspaceStatus.update({ workspaceUuid: uuid }, { mode: 'active' })
|
|
}
|
|
} catch (err) {
|
|
ctx.error('failed to restore workspace', { dataId, err })
|
|
} finally {
|
|
await pipeline?.close()
|
|
await queue.shutdown()
|
|
await workspaceStorage?.close()
|
|
}
|
|
}
|
|
|
|
ctx.info('Successfully restored v6 dump')
|
|
} catch (err: any) {
|
|
ctx.error('Failed to restore v6 dump', { err })
|
|
}
|
|
}
|
|
|
|
export async function restoreTrustedV6Workspace (
|
|
ctx: MeasureMetricsContext,
|
|
accountDB: AccountDB,
|
|
workspace: OldWorkspace,
|
|
accounts: OldAccount[],
|
|
invites: any[],
|
|
backupWsStorage: BackupStorage,
|
|
workspaceStorage: StorageAdapter,
|
|
txes: Tx[],
|
|
dbUrl: string,
|
|
opts?: {
|
|
conflictSuffix?: string
|
|
region?: string
|
|
branding?: string
|
|
force?: boolean
|
|
}
|
|
): Promise<void> {
|
|
const { conflictSuffix, region, branding, force } = opts ?? {}
|
|
// Mapping between <ObjectId, UUID>
|
|
const accountsIdToUuid: Record<string, AccountUuid> = {}
|
|
// Mapping between <email, UUID>
|
|
const accountsEmailToUuid: Record<string, AccountUuid> = {}
|
|
let workspaceUuid: WorkspaceUuid | undefined
|
|
let newWorkspaceUrl: string | undefined
|
|
|
|
ctx.info('Restoring workspace accounts...')
|
|
|
|
let accountsProcessed = 0
|
|
for (const account of accounts) {
|
|
try {
|
|
const accountUuid = await migrateAccount(account, accountDB, false)
|
|
if (accountUuid == null) {
|
|
ctx.warn('Account not restored', account)
|
|
continue
|
|
}
|
|
|
|
accountsIdToUuid[account._id.toString()] = accountUuid
|
|
accountsEmailToUuid[account.email] = accountUuid
|
|
|
|
accountsProcessed++
|
|
if (accountsProcessed % 100 === 0) {
|
|
ctx.info('Processed accounts:', { accountsProcessed })
|
|
}
|
|
} catch (err: any) {
|
|
ctx.error('Failed to restore account', { _id: account._id, email: account.email, err })
|
|
}
|
|
}
|
|
|
|
ctx.info('Total accounts processed:', { accountsProcessed })
|
|
|
|
const oldMode = workspace.mode
|
|
|
|
try {
|
|
// Create workspace with manual-creation mode until it is restored
|
|
;[workspaceUuid, newWorkspaceUrl] =
|
|
(await migrateWorkspace(
|
|
workspace,
|
|
accountDB,
|
|
accountsIdToUuid,
|
|
accountsEmailToUuid,
|
|
false,
|
|
'manual-creation',
|
|
conflictSuffix,
|
|
force !== true,
|
|
region,
|
|
branding
|
|
)) ?? []
|
|
|
|
if (workspaceUuid === undefined) {
|
|
ctx.error('Workspace uuid not set', { workspace: workspace.workspace })
|
|
throw new Error(`Workspace uuid not set ${workspace.workspace}`)
|
|
}
|
|
|
|
if (newWorkspaceUrl == null) {
|
|
ctx.error('Workspace url not set', { workspace: workspace.workspace })
|
|
throw new Error(`Workspace url not set ${workspace.workspace}`)
|
|
}
|
|
|
|
let invitesProcessed = 0
|
|
for (const invite of invites) {
|
|
try {
|
|
if (workspace.workspace !== invite.workspace.name) {
|
|
ctx.error(
|
|
`Invite workspace ${invite.workspace.name} doesn't match workspace being restored ${workspace.workspace}`
|
|
)
|
|
continue
|
|
}
|
|
|
|
const existing = await accountDB.invite.findOne({ migratedFrom: invite._id.toString() })
|
|
if (existing != null) {
|
|
continue
|
|
}
|
|
|
|
const inviteRecord = {
|
|
migratedFrom: invite._id.toString(),
|
|
workspaceUuid,
|
|
expiresOn: invite.exp,
|
|
emailPattern: invite.emailMask,
|
|
remainingUses: invite.limit,
|
|
role: invite.role ?? AccountRole.User
|
|
}
|
|
|
|
await accountDB.invite.insertOne(inviteRecord)
|
|
|
|
invitesProcessed++
|
|
if (invitesProcessed % 100 === 0) {
|
|
ctx.info('Processed invites:', { invitesProcessed })
|
|
}
|
|
} catch (err: any) {
|
|
ctx.error('Failed to restore invite', { _id: invite._id, err })
|
|
}
|
|
}
|
|
|
|
ctx.info('Total invites processed:', { invitesProcessed })
|
|
|
|
const dataId = workspace.workspace
|
|
const url = newWorkspaceUrl
|
|
const uuid = workspaceUuid
|
|
|
|
const wsIds = {
|
|
uuid,
|
|
dataId,
|
|
url
|
|
}
|
|
|
|
const queue = getPlatformQueue('tool', region)
|
|
const wsProducer = queue.getProducer<QueueWorkspaceMessage>(ctx, QueueTopic.Workspace)
|
|
|
|
await wsProducer.send(ctx, uuid, [workspaceEvents.restoring()])
|
|
|
|
let pipeline: Pipeline | undefined
|
|
try {
|
|
pipeline = await createBackupPipeline(ctx, dbUrl, txes, {
|
|
externalStorage: workspaceStorage,
|
|
usePassedCtx: true
|
|
})(ctx, wsIds, createEmptyBroadcastOps(), null)
|
|
if (pipeline === undefined) {
|
|
ctx.error('failed to restore, pipeline is undefined', { dataId })
|
|
return
|
|
}
|
|
await sendTransactorEvent(uuid, 'force-maintenance')
|
|
|
|
await restore(ctx, pipeline, wsIds, backupWsStorage, {
|
|
date: -1,
|
|
merge: false,
|
|
parallel: 1,
|
|
recheck: false
|
|
})
|
|
|
|
await sendTransactorEvent(uuid, 'force-close')
|
|
|
|
ctx.info('workspace restored', { dataId })
|
|
await wsProducer.send(ctx, uuid, [workspaceEvents.restored()])
|
|
|
|
await accountDB.workspaceStatus.update({ workspaceUuid: uuid }, { mode: oldMode })
|
|
} catch (err) {
|
|
ctx.error('failed to restore backup of the workspace', { url, dataId, err })
|
|
} finally {
|
|
await pipeline?.close()
|
|
await queue.shutdown()
|
|
await workspaceStorage?.close()
|
|
}
|
|
} catch (err: any) {
|
|
ctx.error('Failed to restore workspace', { url: workspace.workspaceUrl, workspace: workspace.workspace, err })
|
|
}
|
|
}
|