/* 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();}
}