221 lines
20 KiB
JavaScript
221 lines
20 KiB
JavaScript
/* Resumable, verified migration. No source objects are deleted. */
|
|
const fs = require('node:fs');
|
|
const path = require('node:path');
|
|
const crypto = require('node:crypto');
|
|
const { Readable, Transform } = require('node:stream');
|
|
const SOURCE_HOST = 'demo-sparkai.oss-cn-shanghai.aliyuncs.com';
|
|
const URL_PATTERN = /https?:\/\/(?:demo-sparkai\.oss-cn-shanghai\.aliyuncs\.com\/|s3\.jiansun\.vip\/gongxiangpan\/)[^\s"'<>`\\)\]}|]+/gi;
|
|
function resourceFor(url) {
|
|
const u = new URL(url);
|
|
const isPrimary=u.hostname===SOURCE_HOST,isPsdBackup=u.hostname==='s3.jiansun.vip'&&u.pathname.startsWith('/gongxiangpan/');
|
|
if ((!isPrimary&&!isPsdBackup) || u.pathname === '/') return null;
|
|
for (const name of ['Signature', 'OSSAccessKeyId', 'Expires', 'security-token']) u.searchParams.delete(name);
|
|
u.protocol = 'https:'; u.hash = ''; u.searchParams.sort();
|
|
const identity = u.toString(), sourceKey = decodeURIComponent(u.pathname.slice(isPsdBackup?'/gongxiangpan/'.length:1));
|
|
if (!sourceKey || sourceKey.split('/').some(p => p === '..')) throw Error('Invalid object key');
|
|
const id = crypto.createHash('sha256').update(identity).digest('hex');
|
|
const namespace=isPrimary?'legacy-oss/demo-sparkai/':'legacy-s3/gongxiangpan/';
|
|
const targetKey = u.search ? namespace+'variants/' + id + path.extname(sourceKey) : namespace + sourceKey;
|
|
return { id, identity, sourceHost:u.hostname, sourceKey, targetKey, variant: Boolean(u.search) };
|
|
}
|
|
function urlsIn(value) {
|
|
const found = new Map();
|
|
function walk(v) {
|
|
if (typeof v === 'string') for (const url of v.match(URL_PATTERN) || []) { const r = resourceFor(url); if (r) found.set(r.id, r); }
|
|
else if (Array.isArray(v)) v.forEach(walk);
|
|
else if (v && typeof v === 'object') Object.entries(v).forEach(([k,val])=>{walk(k);walk(val);});
|
|
}
|
|
walk(value); return [...found.values()];
|
|
}
|
|
function replaceVerified(value, verified) {
|
|
if (typeof value === 'string') return value.replace(URL_PATTERN, url => { const r = resourceFor(url); return r && verified.get(r.id)?.targetUrl || url; });
|
|
if (Array.isArray(value)) return value.map(v => replaceVerified(v, verified));
|
|
if (value && typeof value === 'object') return Object.fromEntries(Object.entries(value).map(([k, v]) => [replaceVerified(k, verified), replaceVerified(v, verified)]));
|
|
return value;
|
|
}
|
|
function quote(s) { return '"' + s.replaceAll('"', '""') + '"'; }
|
|
async function rewriteBatch(sql, column, pk, pkType, rows, verified, backup) {
|
|
const isJson=['json','jsonb'].includes(column.dataType);
|
|
const changes=[],originals=[];
|
|
for(const row of rows){
|
|
const next=replaceVerified(row.value,verified);
|
|
if(JSON.stringify(next)===JSON.stringify(row.value))continue;
|
|
originals.push({table:column.tableName,column:column.columnName,pkColumn:pk,pk:row.pk,value:row.value});
|
|
changes.push({pk:String(row.pk),oldvalue:isJson?JSON.stringify(row.value):row.value,newvalue:isJson?JSON.stringify(next):next});
|
|
}
|
|
if(!changes.length)return 0;
|
|
backup(originals);
|
|
const cast=isJson?'::'+column.dataType:'';
|
|
const updated=await sql.unsafe('update '+quote(column.tableName)+' as t set '+quote(column.columnName)+'=v.newvalue'+cast+' from jsonb_to_recordset($1::jsonb) as v(pk text,oldvalue text,newvalue text) where t.'+quote(pk)+'=v.pk::'+pkType+' and t.'+quote(column.columnName)+(isJson?'::jsonb=v.oldvalue::jsonb':'=v.oldvalue')+' returning t.'+quote(pk),[sql.json(changes)]);
|
|
return updated.length;
|
|
}
|
|
module.exports = { resourceFor, urlsIn, replaceVerified, rewriteBatch };
|
|
if (require.main === module) main().catch(e => { console.error('migration_failed', e.name, String(e.message).replace(/https?:\/\/\S+/g, '[URL]').slice(0,240)); process.exitCode = 1; });
|
|
async function main() {
|
|
const root = process.env.MIGRATION_APP_ROOT || process.cwd();
|
|
require(path.join(root, 'node_modules/dotenv')).config({ path: path.join(root, '.env.local') });
|
|
const postgres = require(path.join(root, 'node_modules/postgres'));
|
|
const { S3Client, PutObjectCommand, GetObjectCommand, HeadObjectCommand } = require(path.join(root, 'node_modules/@aws-sdk/client-s3'));
|
|
const sql = postgres(process.env.POSTGRES_URL, { max: 2, prepare: false, connection: { statement_timeout: 600000 } });
|
|
const dir = process.env.MIGRATION_RUN_DIR || '/var/lib/ai-chatbot/aliyun-qiniu-20261014';
|
|
fs.mkdirSync(dir, { recursive: true, mode: 0o700 });
|
|
const journal = path.join(dir, 'verified.jsonl'), verified = new Map();
|
|
if (fs.existsSync(journal)) for (const line of fs.readFileSync(journal,'utf8').split('\n').filter(Boolean)) { const r = JSON.parse(line); verified.set(r.id,r); }
|
|
let sourceClient, targetClient;
|
|
try {
|
|
const configs = await sql`select * from "StorageConfig"`;
|
|
const source = configs.find(c => c.bucket === 'demo-sparkai' && String(c.endpoint).includes('aliyuncs.com'));
|
|
const target = configs.find(c => c.type === 'qiniu-kodo' && c.isEnabled && c.isDefault);
|
|
if (!source || !target || !target.publicUrl) throw Error('Expected source and default Qiniu configuration unavailable');
|
|
function client(c) { return new S3Client({ endpoint: c.endpoint, region: c.region || 'auto', forcePathStyle: true, credentials: { accessKeyId:c.accessKeyId,secretAccessKey:c.secretAccessKey }, requestChecksumCalculation:'WHEN_REQUIRED',responseChecksumValidation:'WHEN_REQUIRED', maxAttempts: 2 }); }
|
|
sourceClient = client(source); targetClient = client(target);
|
|
const publicUrl = key => target.publicUrl.replace(/\/$/,'') + '/' + key.split('/').map(encodeURIComponent).join('/');
|
|
const command = process.argv[2] || 'audit';
|
|
const settings = command==='migrate-psd-extra'?await sql`select id,key,value from "SystemSettings" where key in ('agent_psd_templates','agent_psd_fonts')`:await sql`select id,key,value from "SystemSettings" where value::text like ${'%'+SOURCE_HOST+'/%'}`;
|
|
const settingsResources = urlsIn(settings.map(s => s.value));
|
|
console.log('settings_inventory', JSON.stringify({ rows: settings.length, keys: settings.map(s => s.key), uniqueObjects:settingsResources.length, alreadyVerified:verified.size }));
|
|
if (command === 'audit') return;
|
|
const resources = new Map(settingsResources.map(r => [r.id,r]));
|
|
const scope = command.endsWith('-all') ? 'all' : command==='migrate-psd-extra'?'psd-extra':command==='migrate-recovery'?'recovery':command==='migrate-literals'?'literals':'settings';
|
|
if(command==='migrate-recovery'){
|
|
resources.clear();
|
|
for(const candidate of JSON.parse(fs.readFileSync(process.env.MIGRATION_RESOURCE_FILE,'utf8'))){
|
|
const canonical=resourceFor(candidate.identity);
|
|
if(!canonical||canonical.id!==candidate.id)throw Error('Invalid recovery identity');
|
|
const backup=new URL(candidate.fetchUrl);
|
|
if(!configs.some(c=>c.publicUrl&&backup.origin===new URL(c.publicUrl).origin&&backup.pathname.startsWith(new URL(c.publicUrl).pathname.replace(/\/$/,'')+'/')))throw Error('Recovery source is not a configured storage origin');
|
|
resources.set(canonical.id,{...canonical,fetchUrl:candidate.fetchUrl});
|
|
}
|
|
}
|
|
let columns = [];
|
|
if (scope === 'all') {
|
|
columns = await sql`select c.table_name as "tableName",c.column_name as "columnName",c.data_type as "dataType" from information_schema.columns c join information_schema.tables t on c.table_schema=t.table_schema and c.table_name=t.table_name where c.table_schema='public' and t.table_type='BASE TABLE' and c.is_generated='NEVER' and c.is_updatable='YES' and c.data_type in ('text','character varying','json','jsonb') order by c.table_name,c.ordinal_position`;
|
|
const excluded = new Set(['StorageConfig','StorageMigrationTask','ImageArchiveJob','__drizzle_migrations','SystemSettings','SystemSettingHistory','SystemSettingSnapshot']);
|
|
const tables = new Map();
|
|
const inventoryJournal=path.join(dir,'inventory.jsonl'), doneFile=path.join(dir,'inventory-completed-tables.json');
|
|
const completedTables=new Set(fs.existsSync(doneFile)?JSON.parse(fs.readFileSync(doneFile,'utf8')):[]);
|
|
if(fs.existsSync(inventoryJournal))for(const line of fs.readFileSync(inventoryJournal,'utf8').split('\n').filter(Boolean)){const r=JSON.parse(line);for(const canonical of urlsIn(r.identity))resources.set(canonical.id,canonical);}
|
|
for (const c of columns) if (!excluded.has(c.tableName)) { if (!tables.has(c.tableName)) tables.set(c.tableName,[]); tables.get(c.tableName).push(c); }
|
|
for (const [table, cols] of tables) {
|
|
if(completedTables.has(table)&&process.env.MIGRATION_RESCAN!=='1')continue;
|
|
console.log('scan_start',table);
|
|
const where = cols.map(c => quote(c.columnName)+'::text like $1').join(' or ');
|
|
let matched=0;
|
|
for await (const rows of sql.unsafe('select '+cols.map(c=>quote(c.columnName)).join(',')+' from '+quote(table)+' where '+where,['%'+SOURCE_HOST+'/%']).cursor(50)){const added=[];for (const row of rows) { matched++; for (const r of urlsIn(row)) if(!resources.has(r.id)){resources.set(r.id,r);added.push(r);} }if(added.length)fs.appendFileSync(inventoryJournal,added.map(r=>JSON.stringify(r)).join('\n')+'\n',{mode:0o600});}
|
|
completedTables.add(table);fs.writeFileSync(doneFile,JSON.stringify([...completedTables]),{mode:0o600});
|
|
if (matched) console.log('database_inventory',JSON.stringify({table,rows:matched,uniqueObjects:resources.size}));
|
|
}
|
|
}
|
|
if(scope==='all'||scope==='literals'){
|
|
if(scope==='literals')resources.clear();
|
|
for (const folder of ['app','components','lib']) {
|
|
function walk(d) { for (const e of fs.readdirSync(d,{withFileTypes:true})) { const f=path.join(d,e.name); if(e.isDirectory())walk(f); else if(/\.(tsx?|jsx?)$/.test(f)) for(const r of urlsIn(fs.readFileSync(f,'utf8')))resources.set(r.id,r); } }
|
|
walk(path.join(root,folder));
|
|
}
|
|
}
|
|
fs.writeFileSync(path.join(dir,'plan-'+scope+'.json'),JSON.stringify([...resources.values()]),{mode:0o600});
|
|
if (command === 'inventory-all') return;
|
|
if (!['migrate-settings','migrate-all','migrate-psd-extra','migrate-recovery','migrate-literals','rewrite-all'].includes(command)) throw Error('Unknown command');
|
|
if (scope === 'settings' && !fs.existsSync(path.join(dir,'settings-before.json'))) fs.writeFileSync(path.join(dir,'settings-before.json'),JSON.stringify(settings),{mode:0o600,flag:'wx'});
|
|
const sourceMetadata=new Map(); const metadataFile=path.join(dir,'source-metadata.jsonl');
|
|
if(fs.existsSync(metadataFile))for(const line of fs.readFileSync(metadataFile,'utf8').split('\n').filter(Boolean)){const r=JSON.parse(line);sourceMetadata.set(r.key,r);}
|
|
const base64url = value => Buffer.from(value).toString('base64').replace(/\+/g,'-').replace(/\//g,'_');
|
|
async function cloudCopy(r) {
|
|
const fetchUrl=r.fetchUrl||r.identity;
|
|
const cached=r.fetchUrl||r.variant||new URL(r.identity).hostname!==SOURCE_HOST?null:sourceMetadata.get(r.sourceKey);
|
|
let size=cached?.size,md5=cached?.md5;
|
|
if(!r.variant&&!cached){const head=await fetch(fetchUrl,{method:'HEAD',headers:{'Accept-Encoding':'identity'},signal:AbortSignal.timeout(45000)});if(!head.ok)throw Error('Source HTTP '+head.status);size=Number(head.headers.get('content-length'));md5=String(head.headers.get('etag')||'').replaceAll('"','').toLowerCase();}
|
|
if(r.variant || !size || !/^[a-f0-9]{32}$/.test(md5||'')) {
|
|
const download=await fetch(fetchUrl,{headers:{'Accept-Encoding':'identity'},signal:AbortSignal.timeout(15*60*1000)});
|
|
if(!download.ok)throw Error('Source HTTP '+download.status);
|
|
const hash=crypto.createHash('md5');size=0;for await(const chunk of Readable.fromWeb(download.body)){hash.update(chunk);size+=chunk.length;}md5=hash.digest('hex');
|
|
}
|
|
if(!size)throw Error('Empty source object');
|
|
const requestPath='/fetch/'+base64url(fetchUrl)+'/to/'+base64url(target.bucket+':'+r.targetKey);
|
|
const signature=base64url(crypto.createHmac('sha1',target.secretAccessKey).update(requestPath+'\n').digest());
|
|
const fetchResult=await fetch('https://iovip.qbox.me'+requestPath,{method:'POST',headers:{Authorization:'QBox '+target.accessKeyId+':'+signature,'Content-Type':'application/x-www-form-urlencoded'},signal:AbortSignal.timeout(15*60*1000)});
|
|
if(!fetchResult.ok)throw Error('Qiniu cloud fetch HTTP '+fetchResult.status);
|
|
const info=await fetchResult.json();if(Number(info.fsize)!==size)throw Error('Cloud target size mismatch');
|
|
const targetUrl=publicUrl(r.targetKey);
|
|
const hashResponse=await fetch(targetUrl+'?qhash/md5',{signal:AbortSignal.timeout(45000)});
|
|
if(!hashResponse.ok)throw Error('CDN hash verification HTTP '+hashResponse.status);
|
|
const targetDigest=await hashResponse.json();
|
|
if(targetDigest.hash!==md5 || Number(targetDigest.fsize)!==size)throw Error('Cloud target checksum mismatch');
|
|
const record={...r,targetUrl,size,md5,transport:'qiniu-cloud-fetch',verifiedAt:new Date().toISOString()};fs.appendFileSync(journal,JSON.stringify(record)+'\n',{mode:0o600});verified.set(r.id,record);
|
|
}
|
|
async function copy(r) {
|
|
if (verified.has(r.id)) return;
|
|
let last;
|
|
for (let attempt=1;attempt<=3;attempt++) {
|
|
let input;
|
|
try {
|
|
if(process.env.MIGRATION_TRANSPORT!=='stream'){await cloudCopy(r);return;}
|
|
const timeout=AbortSignal.timeout(15*60*1000);
|
|
const response=await fetch(r.identity,{signal:timeout,headers:{'Accept-Encoding':'identity'}});
|
|
let size,type,disposition;
|
|
if (response.ok) { size=Number(response.headers.get('content-length'));type=response.headers.get('content-type');disposition=response.headers.get('content-disposition'); input=Readable.fromWeb(response.body); }
|
|
else {
|
|
await response.body?.cancel();
|
|
if(r.variant)throw Error('Source variant HTTP '+response.status);
|
|
const get=await sourceClient.send(new GetObjectCommand({Bucket:source.bucket,Key:r.sourceKey}),{abortSignal:timeout});input=get.Body;size=get.ContentLength;type=get.ContentType;disposition=get.ContentDisposition;
|
|
}
|
|
if (!size || size>5*1024**3)throw Error('Unsupported source size '+size);
|
|
const hash=crypto.createHash('sha256');let readBytes=0;
|
|
const hashStream=new Transform({transform(chunk,enc,cb){hash.update(chunk);readBytes+=chunk.length;cb(null,chunk)}});
|
|
input.on('error',e=>hashStream.destroy(e));
|
|
await targetClient.send(new PutObjectCommand({Bucket:target.bucket,Key:r.targetKey,Body:input.pipe(hashStream),ContentLength:size,ContentType:type||'application/octet-stream',ContentDisposition:disposition||undefined,CacheControl:'public, max-age=31536000, immutable',Metadata:{'migration-source-id':r.id}}),{abortSignal:timeout});
|
|
if(readBytes!==size)throw Error('Source size mismatch'); const sha256=hash.digest('hex');
|
|
const get=await targetClient.send(new GetObjectCommand({Bucket:target.bucket,Key:r.targetKey}),{abortSignal:timeout});const targetHash=crypto.createHash('sha256');let storedBytes=0;for await(const chunk of get.Body){targetHash.update(chunk);storedBytes+=chunk.length;}
|
|
if(storedBytes!==size||targetHash.digest('hex')!==sha256)throw Error('Target checksum mismatch');
|
|
const targetUrl=publicUrl(r.targetKey);const head=await fetch(targetUrl,{method:'HEAD',signal:AbortSignal.timeout(30000)});
|
|
if(!head.ok||Number(head.headers.get('content-length'))!==size)throw Error('CDN verification HTTP '+head.status);
|
|
const record={...r,targetUrl,size,sha256,verifiedAt:new Date().toISOString()};fs.appendFileSync(journal,JSON.stringify(record)+'\n',{mode:0o600});verified.set(r.id,record);return;
|
|
} catch(e) { last=e;input?.destroy();if(attempt<3)await new Promise(resolve=>setTimeout(resolve,attempt*1000)); }
|
|
}
|
|
fs.appendFileSync(path.join(dir,'failures-'+scope+'.jsonl'),JSON.stringify({id:r.id,sourceKey:r.sourceKey,name:last.name,message:String(last.message).replace(/https?:\/\/\S+/g,'[URL]').slice(0,240)})+'\n',{mode:0o600});
|
|
}
|
|
const psdIds=new Set(urlsIn(settings.find(s=>s.key==='agent_psd_templates')?.value).map(r=>r.id));
|
|
const list=[...resources.values()].sort((a,b)=>(psdIds.has(a.id)?0:2)-(psdIds.has(b.id)?0:2)||Number(/\.(psd|psb)$/i.test(a.sourceKey))-Number(/\.(psd|psb)$/i.test(b.sourceKey)));
|
|
const pending=list.filter(r=>!verified.has(r.id));
|
|
const concurrency=Math.max(1,Math.min(96,Number(process.env.MIGRATION_CONCURRENCY||64)));
|
|
let cursor=0,done=0;const start=Date.now();
|
|
function progress(){const p={scope,total:list.length,processed:done,verified:verified.size,remaining:list.filter(r=>!verified.has(r.id)).length,elapsedSeconds:Math.round((Date.now()-start)/1000),verifiedBytes:[...verified.values()].reduce((n,r)=>n+r.size,0)};fs.writeFileSync(path.join(dir,'progress-'+scope+'.json'),JSON.stringify(p),{mode:0o600});console.log('copy_progress',JSON.stringify(p));}
|
|
const interval=setInterval(progress,30000);
|
|
try { if(command!=='rewrite-all')await Promise.all(Array.from({length:concurrency},async()=>{while(cursor<pending.length){await copy(pending[cursor++]);done++;if(done%1000===0)progress();}})); } finally {clearInterval(interval);progress();}
|
|
// Recovery copies may finish in a separate process while the primary transfer runs.
|
|
if(fs.existsSync(journal))for(const line of fs.readFileSync(journal,'utf8').split('\n').filter(Boolean)){const r=JSON.parse(line);verified.set(r.id,r);}
|
|
const missing=list.filter(r=>!verified.has(r.id));
|
|
// Only verified addresses are changed; unresolved references keep their original source.
|
|
let changedSettings=0;
|
|
for(const setting of settings){
|
|
for(let attempt=0;attempt<4;attempt++){
|
|
const [current]=await sql`select value from "SystemSettings" where id=${setting.id}`;const next=replaceVerified(current.value,verified);if(JSON.stringify(next)===JSON.stringify(current.value))break;
|
|
const updated=await sql`update "SystemSettings" set value=${sql.json(next)},"updatedAt"=now() where id=${setting.id} and value=${sql.json(current.value)} returning id`;
|
|
if(updated.length){changedSettings++;break;}if(attempt===3)throw Error('Concurrent setting update prevented replacement');
|
|
}
|
|
}
|
|
if(scope==='all'){
|
|
const backupFile=path.join(dir,'database-before.jsonl.gz');
|
|
for(const c of columns){
|
|
// Administrative audit snapshots preserve their original configuration values.
|
|
// User canvas/chat history remains in scope; obsolete configuration backups do not.
|
|
if(['StorageConfig','StorageMigrationTask','ImageArchiveJob','__drizzle_migrations','SystemSettings','SystemSettingHistory','SystemSettingSnapshot'].includes(c.tableName))continue;
|
|
const pks=await sql`select a.attname,a.atttypid::regtype::text as "pkType" from pg_index i join pg_attribute a on a.attrelid=i.indrelid and a.attnum=any(i.indkey) where i.indrelid=${'"'+c.tableName+'"'}::regclass and i.indisprimary`;
|
|
if(pks.length!==1){continue;} const pk=pks[0].attname;
|
|
let count=0;
|
|
const aliasGuard=c.tableName==='ImageArchiveTask'&&c.columnName==='sourceUrl'?" and coalesce(metadata->>'legacyOssMigrationRun','') <> '20261014'":'';
|
|
for await(const rows of sql.unsafe('select '+quote(pk)+' as pk,'+quote(c.columnName)+' as value from '+quote(c.tableName)+' where '+quote(c.columnName)+'::text like $1'+aliasGuard,['%'+SOURCE_HOST+'/%']).cursor(50)){
|
|
// Flush before updating to retain the original data even on interruption.
|
|
count+=await rewriteBatch(sql,c,pk,pks[0].pkType,rows,verified,backups=>fs.appendFileSync(backupFile,require('node:zlib').gzipSync(backups.map(r=>JSON.stringify(r)).join('\n')+'\n',{level:6}),{mode:0o600}));
|
|
}
|
|
if(count)console.log('references_replaced',JSON.stringify({table:c.tableName,column:c.columnName,rows:count}));
|
|
}
|
|
}
|
|
const [migration]=await sql`select id,value from "SystemSettings" where key='oss_image_migration_config'`;
|
|
if(migration){const next={...migration.value,enabled:false,preferAliyunForUploads:false,targetStorageConfigId:target.id};await sql`update "SystemSettings" set value=${sql.json(next)},"updatedAt"=now() where id=${migration.id}`;}
|
|
await sql`update "StorageConfig" set "isEnabled"=false,"isDefault"=false where id=${source.id}`;
|
|
const result={scope,objects:list.length,verified:list.length-missing.length,missing:missing.length,changedSettings,defaultStorage:'qiniu-kodo',sourceRetained:true,completedAt:new Date().toISOString()};fs.writeFileSync(path.join(dir,'result-'+scope+'.json'),JSON.stringify(result,null,2));console.log('migration_result',JSON.stringify(result));
|
|
if(missing.length)process.exitCode=2;
|
|
} finally {sourceClient?.destroy();targetClient?.destroy();await sql.end();}
|
|
}
|