mirror of
https://github.com/xCyanGrizzly/DragonsStash.git
synced 2026-07-22 23:12:02 +00:00
Compare commits
9 Commits
77aeb4cc00
...
04effed825
| Author | SHA1 | Date | |
|---|---|---|---|
| 04effed825 | |||
| c4d9be83bd | |||
| 7d39a13310 | |||
| 18a0efb3d4 | |||
| 2ccc9820cd | |||
| c72b5a4b48 | |||
| 0bdd4ba0cc | |||
| 901f32ff41 | |||
| ff4e150544 |
94
bot/package-lock.json
generated
94
bot/package-lock.json
generated
@@ -12,8 +12,8 @@
|
||||
"@prisma/client": "^7.4.0",
|
||||
"pg": "^8.18.0",
|
||||
"pino": "^9.6.0",
|
||||
"prebuilt-tdlib": "^0.1008050.0",
|
||||
"tdl": "^8.0.0"
|
||||
"prebuilt-tdlib": "^0.1008064.0",
|
||||
"tdl": "^8.1.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/node": "^20",
|
||||
@@ -566,9 +566,9 @@
|
||||
"license": "MIT"
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/darwin-arm64": {
|
||||
"version": "0.1008050.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/darwin-arm64/-/darwin-arm64-0.1008050.0.tgz",
|
||||
"integrity": "sha512-XrWN7M1gfvnzOBRX0YdXVfhSxIDSs/ZJ16QJ0ILDKe+grOFl/cfl7lwB/hK/MlHC6Rev56f5X7xaWnjMh0vktQ==",
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/darwin-arm64/-/darwin-arm64-0.1008064.0.tgz",
|
||||
"integrity": "sha512-Oq5us+o0g68Jag74RIV3LdLkZxQxJMcOdrVbgmyE7Unk+WcifqTb/gZw1rS6BrW+2SX2LNeGY4zQqqBTNDr17Q==",
|
||||
"cpu": [
|
||||
"arm64"
|
||||
],
|
||||
@@ -579,9 +579,9 @@
|
||||
]
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/darwin-x64": {
|
||||
"version": "0.1008050.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/darwin-x64/-/darwin-x64-0.1008050.0.tgz",
|
||||
"integrity": "sha512-a1UfBW0lYx4tUy5viMPtsbqBfBncCAgDu3FPjljfYTHjP8wfkKFxpp5+8wdxhyqdy3QriWaipVtUXQgOeEWMJg==",
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/darwin-x64/-/darwin-x64-0.1008064.0.tgz",
|
||||
"integrity": "sha512-Pz11xjET2Y3uUJKxkWKBc0dmOtlykmBdZ9D6Ahh+EsoLDLIWHm7M91p6nZT396YZ4n2BL+FtDYK65Ae3LDIA5g==",
|
||||
"cpu": [
|
||||
"x64"
|
||||
],
|
||||
@@ -592,9 +592,22 @@
|
||||
]
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/linux-arm64-glibc": {
|
||||
"version": "0.1008050.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/linux-arm64-glibc/-/linux-arm64-glibc-0.1008050.0.tgz",
|
||||
"integrity": "sha512-HRGspdQYzaBkU+W2M8uY5OgOkmgfTkyHkTYan/dn7EE/38QdIFW0YTvmGrl3DoFV2PA+SeJQw0xqK8tMSyHKaA==",
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/linux-arm64-glibc/-/linux-arm64-glibc-0.1008064.0.tgz",
|
||||
"integrity": "sha512-1kML9+RCfTOTWLzxq2klCN962/XwYhd+SGd4BxOwcmvPniYDNRUXtgMi3qRyV/Flola8dchGFrqZJU4kNZNLuQ==",
|
||||
"cpu": [
|
||||
"arm64"
|
||||
],
|
||||
"license": "0BSD",
|
||||
"optional": true,
|
||||
"os": [
|
||||
"linux"
|
||||
]
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/linux-arm64-musl": {
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/linux-arm64-musl/-/linux-arm64-musl-0.1008064.0.tgz",
|
||||
"integrity": "sha512-tN9FJOR8VDfmOoTHMivAqBfQ/d9Bry9T/9cGSTcms3H4ORun/WO5U5zT8VqadAsqjuiQ8Y9HaUqqz65xBDtcgw==",
|
||||
"cpu": [
|
||||
"arm64"
|
||||
],
|
||||
@@ -605,9 +618,9 @@
|
||||
]
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/linux-x64-glibc": {
|
||||
"version": "0.1008050.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/linux-x64-glibc/-/linux-x64-glibc-0.1008050.0.tgz",
|
||||
"integrity": "sha512-Yf6ve3Dzxc66kV1cijFLn7EXKhPN5YHTjtJABEaCR5euetCI2wZp/1uBsXvyYTuFXqQbMfjO3xUCXUIBhLoChw==",
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/linux-x64-glibc/-/linux-x64-glibc-0.1008064.0.tgz",
|
||||
"integrity": "sha512-7fyCp2uk0BdeHKJ9PyQOCditC9vBXeeIjYPAKKBcrkum5bi1e9txy2g5kkGjqwUkN0ntIniS5QfHEyr17Idr9g==",
|
||||
"cpu": [
|
||||
"x64"
|
||||
],
|
||||
@@ -617,10 +630,30 @@
|
||||
"linux"
|
||||
]
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/linux-x64-musl": {
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/linux-x64-musl/-/linux-x64-musl-0.1008064.0.tgz",
|
||||
"integrity": "sha512-e2zRucrRrrK6M04iQWMfwtrts+VvVtyUwtTP1hF2g3a6jW+AHMzFoB9Wu8fWr+vuJflLIQ6sG9r3lI07Q8NenQ==",
|
||||
"cpu": [
|
||||
"x64"
|
||||
],
|
||||
"license": "0BSD",
|
||||
"optional": true,
|
||||
"os": [
|
||||
"linux"
|
||||
]
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/types": {
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/types/-/types-0.1008064.0.tgz",
|
||||
"integrity": "sha512-eqr1+fiHZ+Gj4lwcITzMp6FwPg8UrxlxxaFjhiJRHL9BlbmD2QkCRHac4wW1Sx8Dzwzd7f+xO21Pgi7TBRSwmw==",
|
||||
"license": "0BSD",
|
||||
"optional": true
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/win32-x64": {
|
||||
"version": "0.1008050.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/win32-x64/-/win32-x64-0.1008050.0.tgz",
|
||||
"integrity": "sha512-4v8tU5bodMcLhzrWWXzIzqdHBIpq0wim+7sDmQWQIMy3kDeIzVtpuM+vQjxrGoeH9oWr2WXSRKuj93ld7G5NbQ==",
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/win32-x64/-/win32-x64-0.1008064.0.tgz",
|
||||
"integrity": "sha512-rkacZWexQw52/EUaLAmbsu2+P3C1/AtinlCjfiX07oQAEg3327BCEZqrcY0ER83D8+MMf2pfwMPCDJKytr4hcg==",
|
||||
"cpu": [
|
||||
"x64"
|
||||
],
|
||||
@@ -1669,16 +1702,19 @@
|
||||
}
|
||||
},
|
||||
"node_modules/prebuilt-tdlib": {
|
||||
"version": "0.1008050.0",
|
||||
"resolved": "https://registry.npmjs.org/prebuilt-tdlib/-/prebuilt-tdlib-0.1008050.0.tgz",
|
||||
"integrity": "sha512-CfeQE1rG51d2iC6m72fzrbCW4mqI17ugil9pVurWHtfUJi1Fcn7zadpTzDoUl4oc1dEtKgM7S24DVP67gcl4SQ==",
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/prebuilt-tdlib/-/prebuilt-tdlib-0.1008064.0.tgz",
|
||||
"integrity": "sha512-jJLowKZoH4slXYrkTkKlEgyGsIGv61AWjDZcxxVxJYu21X3kmukGwbCpk4ML99cJp2CwRsD41GCEQBkKJAwCUg==",
|
||||
"license": "MIT",
|
||||
"optionalDependencies": {
|
||||
"@prebuilt-tdlib/darwin-arm64": "0.1008050.0",
|
||||
"@prebuilt-tdlib/darwin-x64": "0.1008050.0",
|
||||
"@prebuilt-tdlib/linux-arm64-glibc": "0.1008050.0",
|
||||
"@prebuilt-tdlib/linux-x64-glibc": "0.1008050.0",
|
||||
"@prebuilt-tdlib/win32-x64": "0.1008050.0"
|
||||
"@prebuilt-tdlib/darwin-arm64": "0.1008064.0",
|
||||
"@prebuilt-tdlib/darwin-x64": "0.1008064.0",
|
||||
"@prebuilt-tdlib/linux-arm64-glibc": "0.1008064.0",
|
||||
"@prebuilt-tdlib/linux-arm64-musl": "0.1008064.0",
|
||||
"@prebuilt-tdlib/linux-x64-glibc": "0.1008064.0",
|
||||
"@prebuilt-tdlib/linux-x64-musl": "0.1008064.0",
|
||||
"@prebuilt-tdlib/types": "0.1008064.0",
|
||||
"@prebuilt-tdlib/win32-x64": "0.1008064.0"
|
||||
}
|
||||
},
|
||||
"node_modules/prisma": {
|
||||
@@ -1971,13 +2007,13 @@
|
||||
"license": "MIT"
|
||||
},
|
||||
"node_modules/tdl": {
|
||||
"version": "8.0.2",
|
||||
"resolved": "https://registry.npmjs.org/tdl/-/tdl-8.0.2.tgz",
|
||||
"integrity": "sha512-KYxlJ4eao7FUu91U1dCDkaHmK70JAyZ1KqitkKqpPC7rxAiXWhaYxddWvt84UxIYoWbgdd0B70FYJ4p/YqpFCA==",
|
||||
"version": "8.1.0",
|
||||
"resolved": "https://registry.npmjs.org/tdl/-/tdl-8.1.0.tgz",
|
||||
"integrity": "sha512-idpw60gjJdiJALQg0+6UbxtJTMxVhzZAgCO6QzL81gqBYCkEFjm9zM9HwTTQGeOaAavw4yRHymR68yUUiCoKrA==",
|
||||
"hasInstallScript": true,
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"debug": "^4.4.0",
|
||||
"debug": "^4.4.3",
|
||||
"node-addon-api": "^7.1.1",
|
||||
"node-gyp-build": "^4.8.4"
|
||||
},
|
||||
|
||||
@@ -13,8 +13,8 @@
|
||||
"@prisma/client": "^7.4.0",
|
||||
"pg": "^8.18.0",
|
||||
"pino": "^9.6.0",
|
||||
"prebuilt-tdlib": "^0.1008050.0",
|
||||
"tdl": "^8.0.0"
|
||||
"prebuilt-tdlib": "^0.1008064.0",
|
||||
"tdl": "^8.1.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/node": "^20",
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
-- AlterTable: capture TDLib's stable per-content identifier for new packages.
|
||||
-- Existing rows are NULL; they fall through to the other dedup checks until
|
||||
-- they're re-encountered organically.
|
||||
ALTER TABLE "packages" ADD COLUMN "remoteUniqueId" TEXT;
|
||||
|
||||
-- CreateIndex: scoped to source channel because we want to dedup
|
||||
-- per-channel (the same file appearing in two different channels is still
|
||||
-- worth indexing twice — they're different ingestion sources).
|
||||
CREATE INDEX "packages_sourceChannelId_remoteUniqueId_idx"
|
||||
ON "packages"("sourceChannelId", "remoteUniqueId");
|
||||
@@ -472,6 +472,11 @@ model Package {
|
||||
sourceChannelId String
|
||||
sourceMessageId BigInt
|
||||
sourceTopicId BigInt?
|
||||
/// TDLib's `remote.unique_id` for the FIRST part's file. Stable across
|
||||
/// reposts of identical content in the same channel — used as the
|
||||
/// strongest pre-download dedup signal (no false positives unlike
|
||||
/// fileName + size matching).
|
||||
remoteUniqueId String?
|
||||
destChannelId String?
|
||||
destMessageId BigInt?
|
||||
destMessageIds BigInt[] @default([])
|
||||
@@ -503,6 +508,7 @@ model Package {
|
||||
@@index([archiveType])
|
||||
@@index([creator])
|
||||
@@index([packageGroupId])
|
||||
@@index([sourceChannelId, remoteUniqueId])
|
||||
@@map("packages")
|
||||
}
|
||||
|
||||
|
||||
94
worker/package-lock.json
generated
94
worker/package-lock.json
generated
@@ -12,8 +12,8 @@
|
||||
"@prisma/client": "^7.4.0",
|
||||
"pg": "^8.18.0",
|
||||
"pino": "^9.6.0",
|
||||
"prebuilt-tdlib": "^0.1008050.0",
|
||||
"tdl": "^8.0.0",
|
||||
"prebuilt-tdlib": "^0.1008064.0",
|
||||
"tdl": "^8.1.0",
|
||||
"yauzl": "^3.2.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
@@ -568,9 +568,9 @@
|
||||
"license": "MIT"
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/darwin-arm64": {
|
||||
"version": "0.1008050.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/darwin-arm64/-/darwin-arm64-0.1008050.0.tgz",
|
||||
"integrity": "sha512-XrWN7M1gfvnzOBRX0YdXVfhSxIDSs/ZJ16QJ0ILDKe+grOFl/cfl7lwB/hK/MlHC6Rev56f5X7xaWnjMh0vktQ==",
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/darwin-arm64/-/darwin-arm64-0.1008064.0.tgz",
|
||||
"integrity": "sha512-Oq5us+o0g68Jag74RIV3LdLkZxQxJMcOdrVbgmyE7Unk+WcifqTb/gZw1rS6BrW+2SX2LNeGY4zQqqBTNDr17Q==",
|
||||
"cpu": [
|
||||
"arm64"
|
||||
],
|
||||
@@ -581,9 +581,9 @@
|
||||
]
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/darwin-x64": {
|
||||
"version": "0.1008050.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/darwin-x64/-/darwin-x64-0.1008050.0.tgz",
|
||||
"integrity": "sha512-a1UfBW0lYx4tUy5viMPtsbqBfBncCAgDu3FPjljfYTHjP8wfkKFxpp5+8wdxhyqdy3QriWaipVtUXQgOeEWMJg==",
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/darwin-x64/-/darwin-x64-0.1008064.0.tgz",
|
||||
"integrity": "sha512-Pz11xjET2Y3uUJKxkWKBc0dmOtlykmBdZ9D6Ahh+EsoLDLIWHm7M91p6nZT396YZ4n2BL+FtDYK65Ae3LDIA5g==",
|
||||
"cpu": [
|
||||
"x64"
|
||||
],
|
||||
@@ -594,9 +594,22 @@
|
||||
]
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/linux-arm64-glibc": {
|
||||
"version": "0.1008050.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/linux-arm64-glibc/-/linux-arm64-glibc-0.1008050.0.tgz",
|
||||
"integrity": "sha512-HRGspdQYzaBkU+W2M8uY5OgOkmgfTkyHkTYan/dn7EE/38QdIFW0YTvmGrl3DoFV2PA+SeJQw0xqK8tMSyHKaA==",
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/linux-arm64-glibc/-/linux-arm64-glibc-0.1008064.0.tgz",
|
||||
"integrity": "sha512-1kML9+RCfTOTWLzxq2klCN962/XwYhd+SGd4BxOwcmvPniYDNRUXtgMi3qRyV/Flola8dchGFrqZJU4kNZNLuQ==",
|
||||
"cpu": [
|
||||
"arm64"
|
||||
],
|
||||
"license": "0BSD",
|
||||
"optional": true,
|
||||
"os": [
|
||||
"linux"
|
||||
]
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/linux-arm64-musl": {
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/linux-arm64-musl/-/linux-arm64-musl-0.1008064.0.tgz",
|
||||
"integrity": "sha512-tN9FJOR8VDfmOoTHMivAqBfQ/d9Bry9T/9cGSTcms3H4ORun/WO5U5zT8VqadAsqjuiQ8Y9HaUqqz65xBDtcgw==",
|
||||
"cpu": [
|
||||
"arm64"
|
||||
],
|
||||
@@ -607,9 +620,9 @@
|
||||
]
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/linux-x64-glibc": {
|
||||
"version": "0.1008050.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/linux-x64-glibc/-/linux-x64-glibc-0.1008050.0.tgz",
|
||||
"integrity": "sha512-Yf6ve3Dzxc66kV1cijFLn7EXKhPN5YHTjtJABEaCR5euetCI2wZp/1uBsXvyYTuFXqQbMfjO3xUCXUIBhLoChw==",
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/linux-x64-glibc/-/linux-x64-glibc-0.1008064.0.tgz",
|
||||
"integrity": "sha512-7fyCp2uk0BdeHKJ9PyQOCditC9vBXeeIjYPAKKBcrkum5bi1e9txy2g5kkGjqwUkN0ntIniS5QfHEyr17Idr9g==",
|
||||
"cpu": [
|
||||
"x64"
|
||||
],
|
||||
@@ -619,10 +632,30 @@
|
||||
"linux"
|
||||
]
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/linux-x64-musl": {
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/linux-x64-musl/-/linux-x64-musl-0.1008064.0.tgz",
|
||||
"integrity": "sha512-e2zRucrRrrK6M04iQWMfwtrts+VvVtyUwtTP1hF2g3a6jW+AHMzFoB9Wu8fWr+vuJflLIQ6sG9r3lI07Q8NenQ==",
|
||||
"cpu": [
|
||||
"x64"
|
||||
],
|
||||
"license": "0BSD",
|
||||
"optional": true,
|
||||
"os": [
|
||||
"linux"
|
||||
]
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/types": {
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/types/-/types-0.1008064.0.tgz",
|
||||
"integrity": "sha512-eqr1+fiHZ+Gj4lwcITzMp6FwPg8UrxlxxaFjhiJRHL9BlbmD2QkCRHac4wW1Sx8Dzwzd7f+xO21Pgi7TBRSwmw==",
|
||||
"license": "0BSD",
|
||||
"optional": true
|
||||
},
|
||||
"node_modules/@prebuilt-tdlib/win32-x64": {
|
||||
"version": "0.1008050.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/win32-x64/-/win32-x64-0.1008050.0.tgz",
|
||||
"integrity": "sha512-4v8tU5bodMcLhzrWWXzIzqdHBIpq0wim+7sDmQWQIMy3kDeIzVtpuM+vQjxrGoeH9oWr2WXSRKuj93ld7G5NbQ==",
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/@prebuilt-tdlib/win32-x64/-/win32-x64-0.1008064.0.tgz",
|
||||
"integrity": "sha512-rkacZWexQw52/EUaLAmbsu2+P3C1/AtinlCjfiX07oQAEg3327BCEZqrcY0ER83D8+MMf2pfwMPCDJKytr4hcg==",
|
||||
"cpu": [
|
||||
"x64"
|
||||
],
|
||||
@@ -1696,16 +1729,19 @@
|
||||
}
|
||||
},
|
||||
"node_modules/prebuilt-tdlib": {
|
||||
"version": "0.1008050.0",
|
||||
"resolved": "https://registry.npmjs.org/prebuilt-tdlib/-/prebuilt-tdlib-0.1008050.0.tgz",
|
||||
"integrity": "sha512-CfeQE1rG51d2iC6m72fzrbCW4mqI17ugil9pVurWHtfUJi1Fcn7zadpTzDoUl4oc1dEtKgM7S24DVP67gcl4SQ==",
|
||||
"version": "0.1008064.0",
|
||||
"resolved": "https://registry.npmjs.org/prebuilt-tdlib/-/prebuilt-tdlib-0.1008064.0.tgz",
|
||||
"integrity": "sha512-jJLowKZoH4slXYrkTkKlEgyGsIGv61AWjDZcxxVxJYu21X3kmukGwbCpk4ML99cJp2CwRsD41GCEQBkKJAwCUg==",
|
||||
"license": "MIT",
|
||||
"optionalDependencies": {
|
||||
"@prebuilt-tdlib/darwin-arm64": "0.1008050.0",
|
||||
"@prebuilt-tdlib/darwin-x64": "0.1008050.0",
|
||||
"@prebuilt-tdlib/linux-arm64-glibc": "0.1008050.0",
|
||||
"@prebuilt-tdlib/linux-x64-glibc": "0.1008050.0",
|
||||
"@prebuilt-tdlib/win32-x64": "0.1008050.0"
|
||||
"@prebuilt-tdlib/darwin-arm64": "0.1008064.0",
|
||||
"@prebuilt-tdlib/darwin-x64": "0.1008064.0",
|
||||
"@prebuilt-tdlib/linux-arm64-glibc": "0.1008064.0",
|
||||
"@prebuilt-tdlib/linux-arm64-musl": "0.1008064.0",
|
||||
"@prebuilt-tdlib/linux-x64-glibc": "0.1008064.0",
|
||||
"@prebuilt-tdlib/linux-x64-musl": "0.1008064.0",
|
||||
"@prebuilt-tdlib/types": "0.1008064.0",
|
||||
"@prebuilt-tdlib/win32-x64": "0.1008064.0"
|
||||
}
|
||||
},
|
||||
"node_modules/prisma": {
|
||||
@@ -1998,13 +2034,13 @@
|
||||
"license": "MIT"
|
||||
},
|
||||
"node_modules/tdl": {
|
||||
"version": "8.0.2",
|
||||
"resolved": "https://registry.npmjs.org/tdl/-/tdl-8.0.2.tgz",
|
||||
"integrity": "sha512-KYxlJ4eao7FUu91U1dCDkaHmK70JAyZ1KqitkKqpPC7rxAiXWhaYxddWvt84UxIYoWbgdd0B70FYJ4p/YqpFCA==",
|
||||
"version": "8.1.0",
|
||||
"resolved": "https://registry.npmjs.org/tdl/-/tdl-8.1.0.tgz",
|
||||
"integrity": "sha512-idpw60gjJdiJALQg0+6UbxtJTMxVhzZAgCO6QzL81gqBYCkEFjm9zM9HwTTQGeOaAavw4yRHymR68yUUiCoKrA==",
|
||||
"hasInstallScript": true,
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"debug": "^4.4.0",
|
||||
"debug": "^4.4.3",
|
||||
"node-addon-api": "^7.1.1",
|
||||
"node-gyp-build": "^4.8.4"
|
||||
},
|
||||
|
||||
@@ -13,8 +13,8 @@
|
||||
"@prisma/client": "^7.4.0",
|
||||
"pg": "^8.18.0",
|
||||
"pino": "^9.6.0",
|
||||
"prebuilt-tdlib": "^0.1008050.0",
|
||||
"tdl": "^8.0.0",
|
||||
"prebuilt-tdlib": "^0.1008064.0",
|
||||
"tdl": "^8.1.0",
|
||||
"yauzl": "^3.2.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
|
||||
93
worker/src/archive/integrity.ts
Normal file
93
worker/src/archive/integrity.ts
Normal file
@@ -0,0 +1,93 @@
|
||||
import { execFile } from "child_process";
|
||||
import { promisify } from "util";
|
||||
import { childLogger } from "../util/logger.js";
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
const log = childLogger("integrity");
|
||||
|
||||
export type IntegrityResult =
|
||||
| { ok: true }
|
||||
| { ok: false; reason: string };
|
||||
|
||||
/**
|
||||
* Test that the archive can be read end-to-end without errors, BEFORE we
|
||||
* spend bandwidth uploading it to the destination channel. Catches:
|
||||
* - Truncated downloads (rare given our size check, but cheap to confirm)
|
||||
* - CRC errors inside the archive
|
||||
* - Bad central directories
|
||||
* - Encrypted archives (we report them as failures rather than upload
|
||||
* a file users can't extract)
|
||||
*
|
||||
* Returns { ok: true } if the archive is intact. Returns
|
||||
* { ok: false, reason } otherwise. Logs at warn level on failure.
|
||||
*
|
||||
* For multipart archives, pass the first part. unzip / unrar / 7z all
|
||||
* auto-discover sibling parts.
|
||||
*
|
||||
* archiveType "DOCUMENT" is a pass-through — there's no container to test.
|
||||
*/
|
||||
export async function testArchiveIntegrity(
|
||||
archiveType: "ZIP" | "RAR" | "SEVEN_Z" | "DOCUMENT",
|
||||
firstPartPath: string
|
||||
): Promise<IntegrityResult> {
|
||||
if (archiveType === "DOCUMENT") {
|
||||
return { ok: true };
|
||||
}
|
||||
|
||||
try {
|
||||
if (archiveType === "ZIP") {
|
||||
// -t = test, -qq = very quiet (errors only)
|
||||
const { stderr } = await execFileAsync("unzip", ["-tqq", firstPartPath], {
|
||||
timeout: 300_000, // 5 min for very large archives
|
||||
maxBuffer: 10 * 1024 * 1024,
|
||||
});
|
||||
if (stderr && stderr.trim()) {
|
||||
return { ok: false, reason: `unzip -t reported: ${stderr.slice(0, 500)}` };
|
||||
}
|
||||
return { ok: true };
|
||||
}
|
||||
|
||||
if (archiveType === "RAR") {
|
||||
const { stdout, stderr } = await execFileAsync("unrar", ["t", firstPartPath], {
|
||||
timeout: 300_000,
|
||||
maxBuffer: 10 * 1024 * 1024,
|
||||
});
|
||||
// unrar uses non-zero exit code on errors, which becomes a throw.
|
||||
// If it succeeds, "All OK" is in stdout.
|
||||
const combined = `${stdout}\n${stderr}`;
|
||||
if (/All OK/i.test(combined)) {
|
||||
return { ok: true };
|
||||
}
|
||||
return { ok: false, reason: `unrar t did not report "All OK": ${combined.slice(-500)}` };
|
||||
}
|
||||
|
||||
if (archiveType === "SEVEN_Z") {
|
||||
const { stdout, stderr } = await execFileAsync("7z", ["t", firstPartPath], {
|
||||
timeout: 300_000,
|
||||
maxBuffer: 10 * 1024 * 1024,
|
||||
});
|
||||
const combined = `${stdout}\n${stderr}`;
|
||||
if (/Everything is Ok/i.test(combined)) {
|
||||
return { ok: true };
|
||||
}
|
||||
return { ok: false, reason: `7z t did not report "Everything is Ok": ${combined.slice(-500)}` };
|
||||
}
|
||||
|
||||
return { ok: false, reason: `Unknown archive type: ${archiveType}` };
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
// execFile throws on non-zero exit. Try to extract the most useful part.
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const stderr = (err as any)?.stderr as string | undefined;
|
||||
const detail = stderr ? `: ${stderr.slice(0, 500)}` : "";
|
||||
|
||||
// Specifically flag encrypted archives so the caller can record a more
|
||||
// specific SkipReason / notification.
|
||||
if (/password|encrypted|need.*password/i.test(`${msg}${detail}`)) {
|
||||
return { ok: false, reason: `Archive is encrypted (password protected): ${msg}${detail}` };
|
||||
}
|
||||
|
||||
log.debug({ err, archiveType, firstPartPath }, "Archive integrity test failed");
|
||||
return { ok: false, reason: `Integrity test failed: ${msg}${detail}` };
|
||||
}
|
||||
}
|
||||
@@ -11,8 +11,11 @@ export interface TelegramMessage {
|
||||
fileSize: bigint;
|
||||
date: Date;
|
||||
mediaAlbumId?: string;
|
||||
replyToMessageId?: bigint; // NEW
|
||||
caption?: string; // NEW
|
||||
replyToMessageId?: bigint;
|
||||
caption?: string;
|
||||
/** TDLib's `remote.unique_id` for the file — stable across reposts of
|
||||
* the exact same content. Empty string if the message didn't expose it. */
|
||||
remoteUniqueId?: string;
|
||||
}
|
||||
|
||||
export interface ArchiveSet {
|
||||
|
||||
@@ -8,83 +8,122 @@ const execFileAsync = promisify(execFile);
|
||||
const log = childLogger("rar-reader");
|
||||
|
||||
/**
|
||||
* Parse output of `unrar l -v <file>` to extract file metadata.
|
||||
* unrar automatically discovers sibling parts when they're co-located.
|
||||
* Parse output of `unrar lt <file>` to extract file metadata.
|
||||
*
|
||||
* `lt` (list technical) emits one block per archived file with key:value
|
||||
* lines — far more reliable than the column-based default `l -v` output,
|
||||
* which has changed format twice across unrar versions.
|
||||
*
|
||||
* unrar automatically discovers sibling multipart files when they're
|
||||
* co-located (e.g. *.part1.rar + *.part2.rar in the same directory).
|
||||
*
|
||||
* Returns [] on any failure (best-effort: ingestion still succeeds with
|
||||
* an empty file list rather than failing the whole archive).
|
||||
*/
|
||||
export async function readRarContents(
|
||||
firstPartPath: string
|
||||
): Promise<FileEntry[]> {
|
||||
try {
|
||||
const { stdout } = await execFileAsync("unrar", ["l", "-v", firstPartPath], {
|
||||
timeout: 30000,
|
||||
maxBuffer: 10 * 1024 * 1024, // 10MB for very large archives
|
||||
const { stdout } = await execFileAsync("unrar", ["lt", firstPartPath], {
|
||||
timeout: 60_000,
|
||||
maxBuffer: 50 * 1024 * 1024, // 50MB for archives with very many files
|
||||
});
|
||||
|
||||
return parseUnrarOutput(stdout);
|
||||
const entries = parseUnrarTechnical(stdout);
|
||||
if (entries.length === 0) {
|
||||
// Log a sample of the output so we can diagnose format changes
|
||||
log.warn(
|
||||
{ file: firstPartPath, sample: stdout.slice(0, 500) },
|
||||
"unrar lt returned no parseable entries"
|
||||
);
|
||||
}
|
||||
return entries;
|
||||
} catch (err) {
|
||||
log.warn({ err, file: firstPartPath }, "Failed to read RAR contents");
|
||||
return []; // Fallback: return empty on error
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse the tabular output of `unrar l -v`.
|
||||
* Parse `unrar lt` output: header followed by per-file key:value blocks
|
||||
* separated by blank lines.
|
||||
*
|
||||
* Example output format:
|
||||
* Archive: test.rar
|
||||
* Details: RAR 5
|
||||
* Example block:
|
||||
*
|
||||
* Attributes Size Packed Ratio Date Time CRC-32 Name
|
||||
* ----------- --------- --------- ----- -------- ----- -------- ----
|
||||
* ...A.... 12345 10234 83% 2024-01-15 10:30 DEADBEEF folder/file.stl
|
||||
* ----------- --------- --------- ----- -------- ----- -------- ----
|
||||
* Name: folder/file.stl
|
||||
* Type: File
|
||||
* Size: 12345
|
||||
* Packed size: 10234
|
||||
* Ratio: 83%
|
||||
* mtime: 2024-01-15 10:30:00,000000000
|
||||
* Attributes: ..A....
|
||||
* CRC32: DEADBEEF
|
||||
* Host OS: Windows
|
||||
* Compression: RAR 5.0(v50) -m3 -md=32M
|
||||
*/
|
||||
function parseUnrarOutput(output: string): FileEntry[] {
|
||||
function parseUnrarTechnical(output: string): FileEntry[] {
|
||||
const entries: FileEntry[] = [];
|
||||
const lines = output.split("\n");
|
||||
// Split into blocks on blank lines, then on each block read key:value pairs.
|
||||
const blocks = output.split(/\r?\n\s*\r?\n/);
|
||||
|
||||
let inFileList = false;
|
||||
let separatorCount = 0;
|
||||
for (const block of blocks) {
|
||||
const fields = parseBlock(block);
|
||||
if (!fields) continue;
|
||||
|
||||
for (const line of lines) {
|
||||
const trimmed = line.trim();
|
||||
// Only File entries (skip Directory, and anything missing the basics)
|
||||
if (fields.type && fields.type.toLowerCase() !== "file") continue;
|
||||
if (!fields.name || fields.size === undefined) continue;
|
||||
|
||||
// Detect separator lines (------- pattern)
|
||||
if (/^-{5,}/.test(trimmed)) {
|
||||
separatorCount++;
|
||||
if (separatorCount === 1) {
|
||||
inFileList = true;
|
||||
} else if (separatorCount >= 2) {
|
||||
inFileList = false;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!inFileList) continue;
|
||||
|
||||
// Parse file entry line
|
||||
// Format: Attributes Size Packed Ratio Date Time CRC Name
|
||||
const match = trimmed.match(
|
||||
/^\S+\s+(\d+)\s+(\d+)\s+\d+%\s+\S+\s+\S+\s+([0-9A-Fa-f]+)\s+(.+)$/
|
||||
);
|
||||
|
||||
if (match) {
|
||||
const [, uncompressedStr, compressedStr, crc32, filePath] = match;
|
||||
|
||||
// Skip directory entries (typically end with / or have size 0 with dir attributes)
|
||||
if (filePath.endsWith("/") || filePath.endsWith("\\")) continue;
|
||||
|
||||
const ext = path.extname(filePath).toLowerCase();
|
||||
entries.push({
|
||||
path: filePath,
|
||||
fileName: path.basename(filePath),
|
||||
extension: ext ? ext.slice(1) : null,
|
||||
compressedSize: BigInt(compressedStr),
|
||||
uncompressedSize: BigInt(uncompressedStr),
|
||||
crc32: crc32.toLowerCase(),
|
||||
});
|
||||
}
|
||||
const filePath = fields.name;
|
||||
const ext = path.extname(filePath).toLowerCase();
|
||||
entries.push({
|
||||
path: filePath,
|
||||
fileName: path.basename(filePath),
|
||||
extension: ext ? ext.slice(1) : null,
|
||||
uncompressedSize: BigInt(fields.size),
|
||||
compressedSize: fields.packedSize !== undefined
|
||||
? BigInt(fields.packedSize)
|
||||
: BigInt(fields.size),
|
||||
crc32: fields.crc32 ? fields.crc32.toLowerCase() : null,
|
||||
});
|
||||
}
|
||||
|
||||
return entries;
|
||||
}
|
||||
|
||||
interface BlockFields {
|
||||
name?: string;
|
||||
type?: string;
|
||||
size?: string;
|
||||
packedSize?: string;
|
||||
crc32?: string;
|
||||
}
|
||||
|
||||
function parseBlock(block: string): BlockFields | null {
|
||||
// Skip the archive-header block (contains "Archive:" / "Details:" lines
|
||||
// and lacks a Name field).
|
||||
if (!/^\s*Name:/m.test(block)) return null;
|
||||
|
||||
const fields: BlockFields = {};
|
||||
const lines = block.split(/\r?\n/);
|
||||
|
||||
for (const line of lines) {
|
||||
// Match " key: value" with arbitrary leading whitespace and a multi-word
|
||||
// key (e.g. "Packed size", "Host OS").
|
||||
const m = line.match(/^\s*([A-Za-z][A-Za-z0-9 ]*?)\s*:\s*(.*)$/);
|
||||
if (!m) continue;
|
||||
const key = m[1].trim().toLowerCase();
|
||||
const value = m[2].trim();
|
||||
if (key === "name") fields.name = value;
|
||||
else if (key === "type") fields.type = value;
|
||||
else if (key === "size") fields.size = value;
|
||||
else if (key === "packed size") fields.packedSize = value;
|
||||
else if (key === "crc32" || key === "blake2sp" || key === "checksum") {
|
||||
// unrar may report BLAKE2sp for newer archives instead of CRC32.
|
||||
// Either way we just store it as a hex string in our crc32 field.
|
||||
fields.crc32 = value;
|
||||
}
|
||||
}
|
||||
|
||||
return fields;
|
||||
}
|
||||
|
||||
58
worker/src/archive/slicer-tags.ts
Normal file
58
worker/src/archive/slicer-tags.ts
Normal file
@@ -0,0 +1,58 @@
|
||||
import type { FileEntry } from "./zip-reader.js";
|
||||
|
||||
/**
|
||||
* Mapping from file extensions to slicer tags. Each tag groups a family of
|
||||
* extensions that mean the same thing for end users — "this archive contains
|
||||
* files I can open in <slicer>".
|
||||
*
|
||||
* Extensions are matched case-insensitively without the leading dot.
|
||||
*/
|
||||
const SLICER_EXTENSION_MAP: Record<string, string> = {
|
||||
// Lychee Slicer
|
||||
lys: "lychee",
|
||||
lyt: "lychee",
|
||||
lyc: "lychee",
|
||||
|
||||
// ChituBox / Anycubic / Phrozen / Elegoo (resin printers)
|
||||
chitubox: "chitubox",
|
||||
ctb: "chitubox",
|
||||
cbddlp: "chitubox",
|
||||
|
||||
// Anycubic Photon family
|
||||
photon: "anycubic",
|
||||
pwmo: "anycubic",
|
||||
pwmx: "anycubic",
|
||||
pwmb: "anycubic",
|
||||
pwma: "anycubic",
|
||||
pws: "anycubic",
|
||||
pwsq: "anycubic",
|
||||
phz: "anycubic",
|
||||
|
||||
// Bambu / Prusa
|
||||
"3mf": "bambu",
|
||||
bgcode: "bambu",
|
||||
|
||||
// FDM gcode (generic)
|
||||
gcode: "fdm",
|
||||
|
||||
// Mango / generic resin formats sometimes seen in releases
|
||||
mfp: "mango",
|
||||
mfpv: "mango",
|
||||
osla: "mango",
|
||||
};
|
||||
|
||||
/**
|
||||
* Derive a deduplicated list of slicer tags from an archive's file listing.
|
||||
* Returns an empty array if no recognised slicer-specific files are present
|
||||
* (e.g., the archive is just STLs without pre-supports).
|
||||
*/
|
||||
export function extractSlicerTags(entries: FileEntry[]): string[] {
|
||||
const tags = new Set<string>();
|
||||
for (const entry of entries) {
|
||||
if (!entry.extension) continue;
|
||||
const ext = entry.extension.toLowerCase();
|
||||
const tag = SLICER_EXTENSION_MAP[ext];
|
||||
if (tag) tags.add(tag);
|
||||
}
|
||||
return [...tags].sort();
|
||||
}
|
||||
343
worker/src/backfill.ts
Normal file
343
worker/src/backfill.ts
Normal file
@@ -0,0 +1,343 @@
|
||||
import path from "path";
|
||||
import { mkdir, rm } from "fs/promises";
|
||||
import { db } from "./db/client.js";
|
||||
import { config } from "./util/config.js";
|
||||
import { childLogger } from "./util/logger.js";
|
||||
import { withTdlibMutex } from "./util/mutex.js";
|
||||
import { createTdlibClient, closeTdlibClient } from "./tdlib/client.js";
|
||||
import { downloadFile } from "./tdlib/download.js";
|
||||
import { getActiveAccounts } from "./db/queries.js";
|
||||
import { readZipCentralDirectory } from "./archive/zip-reader.js";
|
||||
import { readRarContents } from "./archive/rar-reader.js";
|
||||
import { read7zContents } from "./archive/sevenz-reader.js";
|
||||
import { extractSlicerTags } from "./archive/slicer-tags.js";
|
||||
import type { FileEntry } from "./archive/zip-reader.js";
|
||||
|
||||
const log = childLogger("backfill");
|
||||
|
||||
/**
|
||||
* Re-extract file listings for Packages whose fileCount is 0 — usually
|
||||
* caused by historical bugs in the archive readers (e.g. the RAR parser
|
||||
* that silently returned [] for every archive before 0bdd4ba).
|
||||
*
|
||||
* For each candidate Package:
|
||||
* 1. Download all destMessageIds from the destination channel
|
||||
* 2. Run the appropriate reader (ZIP / RAR / 7Z) on the assembled files
|
||||
* 3. Insert PackageFile rows + update Package.fileCount
|
||||
* 4. Clean up the temp files
|
||||
*
|
||||
* Triggered via pg_notify "backfill_filelists" with optional payload
|
||||
* `{"limit": N, "archiveType": "RAR"}` — both fields optional, defaults
|
||||
* are limit=100, archiveType=any.
|
||||
*/
|
||||
export async function processBackfillRequest(payloadJson: string): Promise<void> {
|
||||
let limit = 100;
|
||||
let archiveTypeFilter: "ZIP" | "RAR" | "SEVEN_Z" | undefined;
|
||||
try {
|
||||
const parsed = JSON.parse(payloadJson) as { limit?: number; archiveType?: string };
|
||||
if (typeof parsed.limit === "number" && parsed.limit > 0) limit = parsed.limit;
|
||||
if (parsed.archiveType === "ZIP" || parsed.archiveType === "RAR" || parsed.archiveType === "SEVEN_Z") {
|
||||
archiveTypeFilter = parsed.archiveType;
|
||||
}
|
||||
} catch {
|
||||
// Empty / invalid payload — use defaults
|
||||
}
|
||||
|
||||
const candidates = await db.package.findMany({
|
||||
where: {
|
||||
fileCount: 0,
|
||||
destChannelId: { not: null },
|
||||
destMessageId: { not: null },
|
||||
archiveType: archiveTypeFilter
|
||||
? archiveTypeFilter
|
||||
: { in: ["ZIP", "RAR", "SEVEN_Z"] },
|
||||
},
|
||||
select: {
|
||||
id: true,
|
||||
fileName: true,
|
||||
fileSize: true,
|
||||
archiveType: true,
|
||||
destChannelId: true,
|
||||
destMessageId: true,
|
||||
destMessageIds: true,
|
||||
isMultipart: true,
|
||||
partCount: true,
|
||||
},
|
||||
orderBy: { createdAt: "asc" },
|
||||
take: limit,
|
||||
});
|
||||
|
||||
if (candidates.length === 0) {
|
||||
log.info({ archiveTypeFilter }, "Backfill: no candidates with fileCount=0");
|
||||
return;
|
||||
}
|
||||
|
||||
log.info(
|
||||
{ count: candidates.length, archiveTypeFilter },
|
||||
"Backfill: starting batch"
|
||||
);
|
||||
|
||||
const accounts = await getActiveAccounts();
|
||||
if (accounts.length === 0) {
|
||||
log.warn("Backfill: no authenticated accounts — aborting");
|
||||
return;
|
||||
}
|
||||
|
||||
// Prefer the Premium account if available (faster downloads, larger files)
|
||||
const account = accounts.find((a) => a.isPremium) ?? accounts[0];
|
||||
|
||||
await withTdlibMutex(account.phone, "backfill", async () => {
|
||||
const { client } = await createTdlibClient({ id: account.id, phone: account.phone });
|
||||
|
||||
try {
|
||||
// Load chats so TDLib knows about the destination chat
|
||||
try {
|
||||
await client.invoke({
|
||||
_: "getChats",
|
||||
chat_list: { _: "chatListMain" },
|
||||
limit: 1000,
|
||||
});
|
||||
} catch {
|
||||
// May already be loaded
|
||||
}
|
||||
|
||||
let processed = 0;
|
||||
let succeeded = 0;
|
||||
let failed = 0;
|
||||
|
||||
for (const pkg of candidates) {
|
||||
processed++;
|
||||
const ctx = { packageId: pkg.id, fileName: pkg.fileName };
|
||||
|
||||
try {
|
||||
await processOnePackage(client, pkg, ctx);
|
||||
succeeded++;
|
||||
} catch (err) {
|
||||
failed++;
|
||||
log.warn({ err, ...ctx }, "Backfill failed for package");
|
||||
}
|
||||
}
|
||||
|
||||
log.info(
|
||||
{ processed, succeeded, failed, archiveTypeFilter },
|
||||
"Backfill batch complete"
|
||||
);
|
||||
} finally {
|
||||
await closeTdlibClient(client).catch(() => {});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
interface BackfillPackage {
|
||||
id: string;
|
||||
fileName: string;
|
||||
fileSize: bigint;
|
||||
archiveType: "ZIP" | "RAR" | "SEVEN_Z" | "DOCUMENT" | string;
|
||||
destChannelId: string | null;
|
||||
destMessageId: bigint | null;
|
||||
destMessageIds: bigint[];
|
||||
isMultipart: boolean;
|
||||
partCount: number;
|
||||
}
|
||||
|
||||
async function processOnePackage(
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
client: any,
|
||||
pkg: BackfillPackage,
|
||||
ctx: { packageId: string; fileName: string }
|
||||
): Promise<void> {
|
||||
if (!pkg.destChannelId || !pkg.destMessageId) {
|
||||
log.debug(ctx, "Skipping: no destination channel/message");
|
||||
return;
|
||||
}
|
||||
|
||||
// Look up the destination channel's Telegram ID
|
||||
const destChannel = await db.telegramChannel.findUnique({
|
||||
where: { id: pkg.destChannelId },
|
||||
select: { telegramId: true },
|
||||
});
|
||||
if (!destChannel) {
|
||||
throw new Error("Destination channel not found in DB");
|
||||
}
|
||||
const chatId = Number(destChannel.telegramId);
|
||||
|
||||
// Resolve which message IDs to download. The Package may carry a
|
||||
// single destMessageId or multiple destMessageIds (for multipart).
|
||||
const messageIds: bigint[] =
|
||||
pkg.destMessageIds.length > 0
|
||||
? pkg.destMessageIds
|
||||
: pkg.destMessageId
|
||||
? [pkg.destMessageId]
|
||||
: [];
|
||||
|
||||
if (messageIds.length === 0) {
|
||||
throw new Error("Package has no destination message IDs");
|
||||
}
|
||||
|
||||
const tempDir = path.join(config.tempDir, `backfill_${pkg.id}`);
|
||||
await mkdir(tempDir, { recursive: true });
|
||||
|
||||
try {
|
||||
const partPaths: string[] = [];
|
||||
|
||||
for (let i = 0; i < messageIds.length; i++) {
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const message = (await client.invoke({
|
||||
_: "getMessage",
|
||||
chat_id: chatId,
|
||||
message_id: Number(messageIds[i]),
|
||||
})) as unknown as {
|
||||
content?: { document?: { file_name?: string; document?: { id: number; size: number } } };
|
||||
};
|
||||
|
||||
const doc = message?.content?.document;
|
||||
if (!doc?.document?.id) {
|
||||
throw new Error(`Destination message ${messageIds[i]} has no document`);
|
||||
}
|
||||
|
||||
const fileId = String(doc.document.id);
|
||||
const fileName = doc.file_name ?? `${pkg.id}.part${i + 1}`;
|
||||
const localPath = path.join(tempDir, fileName);
|
||||
|
||||
await downloadFile(
|
||||
client,
|
||||
fileId,
|
||||
localPath,
|
||||
BigInt(doc.document.size),
|
||||
fileName
|
||||
);
|
||||
|
||||
partPaths.push(localPath);
|
||||
}
|
||||
|
||||
// Run the appropriate reader on the assembled file(s)
|
||||
let entries: FileEntry[] = [];
|
||||
if (pkg.archiveType === "ZIP") {
|
||||
entries = await readZipCentralDirectory(partPaths);
|
||||
} else if (pkg.archiveType === "RAR") {
|
||||
// unrar auto-discovers sibling parts when in the same directory
|
||||
entries = await readRarContents(partPaths[0]);
|
||||
} else if (pkg.archiveType === "SEVEN_Z") {
|
||||
entries = await read7zContents(partPaths[0]);
|
||||
} else {
|
||||
log.debug({ ...ctx, archiveType: pkg.archiveType }, "Skipping unsupported archive type");
|
||||
return;
|
||||
}
|
||||
|
||||
if (entries.length === 0) {
|
||||
log.warn(ctx, "Reader returned 0 entries — archive may be encrypted or corrupt");
|
||||
return;
|
||||
}
|
||||
|
||||
// Also derive slicer tags from the file list so the backfilled packages
|
||||
// gain the same search/filter context as newly-ingested ones.
|
||||
const slicerTags = extractSlicerTags(entries);
|
||||
|
||||
// Write everything in a single transaction so a partial backfill never
|
||||
// leaves the Package half-indexed.
|
||||
await db.$transaction(async (tx) => {
|
||||
// Re-check fileCount inside the transaction: another worker might
|
||||
// have backfilled this package between our read and write.
|
||||
const current = await tx.package.findUnique({
|
||||
where: { id: pkg.id },
|
||||
select: { fileCount: true, tags: true },
|
||||
});
|
||||
if (current && current.fileCount > 0) {
|
||||
log.debug({ ...ctx, existingFileCount: current.fileCount }, "Already backfilled by another worker — skipping");
|
||||
return;
|
||||
}
|
||||
|
||||
await tx.packageFile.deleteMany({ where: { packageId: pkg.id } });
|
||||
await tx.packageFile.createMany({
|
||||
data: entries.map((e) => ({
|
||||
packageId: pkg.id,
|
||||
path: e.path,
|
||||
fileName: e.fileName,
|
||||
extension: e.extension,
|
||||
compressedSize: e.compressedSize,
|
||||
uncompressedSize: e.uncompressedSize,
|
||||
crc32: e.crc32,
|
||||
})),
|
||||
});
|
||||
|
||||
// Merge slicer tags with whatever's already on the Package (preserve
|
||||
// channel category, manual tags, etc.).
|
||||
const existingTags = current?.tags ?? [];
|
||||
const mergedTags = [...new Set([...existingTags, ...slicerTags])];
|
||||
|
||||
await tx.package.update({
|
||||
where: { id: pkg.id },
|
||||
data: { fileCount: entries.length, tags: mergedTags },
|
||||
});
|
||||
});
|
||||
|
||||
log.info({ ...ctx, fileCount: entries.length }, "Backfilled file list");
|
||||
} finally {
|
||||
await rm(tempDir, { recursive: true, force: true }).catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Cheap pure-DB backfill: walk Packages that already have PackageFile rows
|
||||
* but no slicer tags, recompute the tags from their extensions, and merge
|
||||
* with the existing tag list. No downloads, no TDLib.
|
||||
*
|
||||
* Trigger:
|
||||
* SELECT pg_notify('backfill_slicer_tags', '{"limit":1000}');
|
||||
*/
|
||||
export async function processSlicerTagBackfill(payloadJson: string): Promise<void> {
|
||||
let limit = 1000;
|
||||
try {
|
||||
const parsed = JSON.parse(payloadJson) as { limit?: number };
|
||||
if (typeof parsed.limit === "number" && parsed.limit > 0) limit = parsed.limit;
|
||||
} catch {
|
||||
// Use default
|
||||
}
|
||||
|
||||
// KNOWN_TAGS = the slicer tags we know how to derive. A Package missing
|
||||
// all of these is a candidate for recompute. extractSlicerTags is safe
|
||||
// to run on every package (returns [] for archives with no slicer files),
|
||||
// but filtering up-front avoids walking the entire DB.
|
||||
const KNOWN_TAGS = ["lychee", "chitubox", "anycubic", "bambu", "fdm", "mango"];
|
||||
|
||||
const candidates = await db.package.findMany({
|
||||
where: {
|
||||
fileCount: { gt: 0 },
|
||||
NOT: { tags: { hasSome: KNOWN_TAGS } },
|
||||
},
|
||||
select: {
|
||||
id: true,
|
||||
tags: true,
|
||||
files: { select: { extension: true } },
|
||||
},
|
||||
orderBy: { createdAt: "asc" },
|
||||
take: limit,
|
||||
});
|
||||
|
||||
if (candidates.length === 0) {
|
||||
log.info("Slicer tag backfill: no candidates");
|
||||
return;
|
||||
}
|
||||
|
||||
log.info({ count: candidates.length }, "Slicer tag backfill: starting");
|
||||
|
||||
let updated = 0;
|
||||
for (const pkg of candidates) {
|
||||
const fileEntries = pkg.files.map((f) => ({
|
||||
path: "",
|
||||
fileName: "",
|
||||
extension: f.extension,
|
||||
compressedSize: 0n,
|
||||
uncompressedSize: 0n,
|
||||
crc32: null as string | null,
|
||||
}));
|
||||
const slicerTags = extractSlicerTags(fileEntries);
|
||||
if (slicerTags.length === 0) continue;
|
||||
const merged = [...new Set([...pkg.tags, ...slicerTags])];
|
||||
if (merged.length === pkg.tags.length) continue;
|
||||
await db.package.update({ where: { id: pkg.id }, data: { tags: merged } });
|
||||
updated++;
|
||||
}
|
||||
|
||||
log.info({ candidates: candidates.length, updated }, "Slicer tag backfill: done");
|
||||
}
|
||||
@@ -82,6 +82,8 @@ export interface CreatePackageStubInput {
|
||||
sourceChannelId: string;
|
||||
sourceMessageId: bigint;
|
||||
sourceTopicId?: bigint | null;
|
||||
/** TDLib remote.unique_id of the first part — for future dedup. */
|
||||
remoteUniqueId?: string | null;
|
||||
destChannelId: string;
|
||||
destMessageId: bigint;
|
||||
destMessageIds: bigint[];
|
||||
@@ -111,6 +113,7 @@ export async function createPackageStub(
|
||||
sourceChannelId: input.sourceChannelId,
|
||||
sourceMessageId: input.sourceMessageId,
|
||||
sourceTopicId: input.sourceTopicId ?? undefined,
|
||||
remoteUniqueId: input.remoteUniqueId ?? undefined,
|
||||
destChannelId: input.destChannelId,
|
||||
destMessageId: input.destMessageId,
|
||||
destMessageIds: input.destMessageIds,
|
||||
@@ -189,6 +192,73 @@ export async function packageExistsBySourceMessage(
|
||||
return pkg !== null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Strongest pre-download dedup signal: a Package in this channel already
|
||||
* has a matching TDLib remote.unique_id. The unique_id is stable across
|
||||
* reposts of the exact same file content, so a hit is a guaranteed
|
||||
* (lossless) duplicate. No false positives.
|
||||
*
|
||||
* Falls back to the older findRepostedPackage (name + size) for packages
|
||||
* that were ingested before we started capturing remote.unique_id.
|
||||
*/
|
||||
export async function findPackageByRemoteUniqueId(
|
||||
sourceChannelId: string,
|
||||
remoteUniqueId: string
|
||||
): Promise<{
|
||||
id: string;
|
||||
destMessageId: bigint | null;
|
||||
sourceTopicId: bigint | null;
|
||||
} | null> {
|
||||
return db.package.findFirst({
|
||||
where: {
|
||||
sourceChannelId,
|
||||
remoteUniqueId,
|
||||
destMessageId: { not: null },
|
||||
},
|
||||
orderBy: { sourceTopicId: { sort: "desc", nulls: "last" } },
|
||||
select: { id: true, destMessageId: true, sourceTopicId: true },
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Detect a likely repost: same source channel + same fileName + same total
|
||||
* fileSize already exists with destMessageId set. Used to skip downloads
|
||||
* when the channel admin re-posts the same file under a new message ID
|
||||
* (which `packageExistsBySourceMessage` cannot catch because the message ID
|
||||
* is different).
|
||||
*
|
||||
* Returns the existing package's destMessageId for logging/observability,
|
||||
* or null if no match. Approximate: same name + same total size is an
|
||||
* extremely strong signal that it's the same content, but theoretically
|
||||
* two unrelated files could collide. If that ever happens, the new file
|
||||
* gets treated as a duplicate and is lost; the user can manually re-link
|
||||
* via the UI by removing the existing Package.
|
||||
*/
|
||||
export async function findRepostedPackage(
|
||||
sourceChannelId: string,
|
||||
fileName: string,
|
||||
fileSize: bigint
|
||||
): Promise<{
|
||||
id: string;
|
||||
destMessageId: bigint | null;
|
||||
sourceTopicId: bigint | null;
|
||||
} | null> {
|
||||
return db.package.findFirst({
|
||||
where: {
|
||||
sourceChannelId,
|
||||
fileName,
|
||||
fileSize,
|
||||
destMessageId: { not: null },
|
||||
},
|
||||
// Prefer the existing Package with the most specific (non-NULL)
|
||||
// sourceTopicId, so when the user is re-scanning and the file already
|
||||
// exists in a specific topic, the audit log shows the most informative
|
||||
// match. NULLS LAST in DESC order achieves that for any non-null IDs.
|
||||
orderBy: { sourceTopicId: { sort: "desc", nulls: "last" } },
|
||||
select: { id: true, destMessageId: true, sourceTopicId: true },
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete orphaned Package rows that have the same content hash but never
|
||||
* completed the upload (destMessageId is null). Called before creating a
|
||||
@@ -672,6 +742,63 @@ export async function deleteSkippedPackage(
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Find SkippedPackages for a given account+channel that are still eligible
|
||||
* for auto-retry (attemptCount below the cap). Used at the start of a scan
|
||||
* to pull the watermark back so we don't strand failed messages forever
|
||||
* after the watermark has advanced past them.
|
||||
*
|
||||
* For non-forum channels, pass `topicId: null` to get rows with NULL topic.
|
||||
* For forum channels, pass the topic ID to scope to that topic only.
|
||||
*/
|
||||
export async function getRetryableSkippedMessageIds(args: {
|
||||
accountId: string;
|
||||
sourceChannelId: string;
|
||||
topicId: bigint | null;
|
||||
cap: number;
|
||||
}): Promise<bigint[]> {
|
||||
const rows = await db.skippedPackage.findMany({
|
||||
where: {
|
||||
accountId: args.accountId,
|
||||
sourceChannelId: args.sourceChannelId,
|
||||
sourceTopicId: args.topicId,
|
||||
attemptCount: { lt: args.cap },
|
||||
},
|
||||
select: { sourceMessageId: true },
|
||||
orderBy: { sourceMessageId: "asc" },
|
||||
});
|
||||
return rows.map((r) => r.sourceMessageId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Update a Package's source topic when a more specific topic context is
|
||||
* discovered for the same content. Used when findRepostedPackage matches
|
||||
* an existing Package whose topic is less specific (e.g., "General") than
|
||||
* the topic we just encountered the file in.
|
||||
*
|
||||
* Also updates the creator if the new topic name is more informative than
|
||||
* the existing creator (i.e., the existing creator was derived from a
|
||||
* less-specific topic name like "General").
|
||||
*/
|
||||
export async function updatePackageTopicContext(
|
||||
packageId: string,
|
||||
newTopicId: bigint,
|
||||
newTopicName: string | null
|
||||
): Promise<void> {
|
||||
await db.package.update({
|
||||
where: { id: packageId },
|
||||
data: {
|
||||
sourceTopicId: newTopicId,
|
||||
// Only overwrite creator if the new topic name is meaningful (non-empty,
|
||||
// non-General). Keeps explicit creator values from filename or admin
|
||||
// input intact.
|
||||
...(newTopicName && newTopicName !== "General"
|
||||
? { creator: newTopicName }
|
||||
: {}),
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
export async function createOrFindPackageGroup(input: {
|
||||
mediaAlbumId: string;
|
||||
sourceChannelId: string;
|
||||
|
||||
@@ -6,6 +6,7 @@ import { processFetchRequest } from "./worker.js";
|
||||
import { processExtractRequest } from "./extract-listener.js";
|
||||
import { rebuildPackageDatabase } from "./rebuild.js";
|
||||
import { processManualUpload } from "./manual-upload.js";
|
||||
import { processBackfillRequest, processSlicerTagBackfill } from "./backfill.js";
|
||||
import { generateInviteLink, createSupergroup, searchPublicChat } from "./tdlib/chats.js";
|
||||
import { createTdlibClient, closeTdlibClient } from "./tdlib/client.js";
|
||||
import { triggerImmediateCycle } from "./scheduler.js";
|
||||
@@ -58,6 +59,8 @@ async function connectListener(): Promise<void> {
|
||||
await pgClient.query("LISTEN archive_extract");
|
||||
await pgClient.query("LISTEN rebuild_packages");
|
||||
await pgClient.query("LISTEN manual_upload");
|
||||
await pgClient.query("LISTEN backfill_filelists");
|
||||
await pgClient.query("LISTEN backfill_slicer_tags");
|
||||
|
||||
pgClient.on("notification", (msg) => {
|
||||
if (msg.channel === "channel_fetch" && msg.payload) {
|
||||
@@ -76,6 +79,10 @@ async function connectListener(): Promise<void> {
|
||||
handleRebuildPackages(msg.payload);
|
||||
} else if (msg.channel === "manual_upload" && msg.payload) {
|
||||
handleManualUpload(msg.payload);
|
||||
} else if (msg.channel === "backfill_filelists") {
|
||||
handleBackfillFilelists(msg.payload ?? "{}");
|
||||
} else if (msg.channel === "backfill_slicer_tags") {
|
||||
handleBackfillSlicerTags(msg.payload ?? "{}");
|
||||
}
|
||||
});
|
||||
|
||||
@@ -101,7 +108,7 @@ async function connectListener(): Promise<void> {
|
||||
}
|
||||
});
|
||||
|
||||
log.info("Fetch listener started (channel_fetch, generate_invite, create_destination, ingestion_trigger, join_channel, archive_extract, rebuild_packages, manual_upload)");
|
||||
log.info("Fetch listener started (channel_fetch, generate_invite, create_destination, ingestion_trigger, join_channel, archive_extract, rebuild_packages, manual_upload, backfill_filelists, backfill_slicer_tags)");
|
||||
} catch (err) {
|
||||
log.error({ err }, "Failed to start fetch listener — retrying");
|
||||
scheduleReconnect();
|
||||
@@ -528,3 +535,30 @@ function handleManualUpload(uploadId: string): void {
|
||||
.then(() => processManualUpload(uploadId))
|
||||
.catch((err) => log.error({ err, uploadId }, "Manual upload processing failed"));
|
||||
}
|
||||
|
||||
// ── Backfill file-list handler ──
|
||||
//
|
||||
// Trigger via:
|
||||
// SELECT pg_notify('backfill_filelists', '{"limit":50,"archiveType":"RAR"}');
|
||||
//
|
||||
// Both fields are optional. archiveType filters to one of ZIP/RAR/SEVEN_Z.
|
||||
// Default limit is 100. The handler queues so multiple notifications run
|
||||
// sequentially (no concurrent TDLib downloads competing for the mutex).
|
||||
function handleBackfillFilelists(payload: string): void {
|
||||
fetchQueue = fetchQueue
|
||||
.then(() => processBackfillRequest(payload))
|
||||
.catch((err) => log.error({ err, payload }, "Backfill request failed"));
|
||||
}
|
||||
|
||||
// ── Slicer tag backfill handler ──
|
||||
//
|
||||
// Trigger:
|
||||
// SELECT pg_notify('backfill_slicer_tags', '{"limit":1000}');
|
||||
//
|
||||
// Pure-DB pass over Packages that have file lists but no slicer tags.
|
||||
// No downloads, no TDLib involvement — fast and safe.
|
||||
function handleBackfillSlicerTags(payload: string): void {
|
||||
fetchQueue = fetchQueue
|
||||
.then(() => processSlicerTagBackfill(payload))
|
||||
.catch((err) => log.error({ err, payload }, "Slicer tag backfill failed"));
|
||||
}
|
||||
|
||||
@@ -78,18 +78,31 @@ export async function recoverIncompleteUploads(): Promise<void> {
|
||||
|
||||
let resetCount = 0;
|
||||
let verifiedCount = 0;
|
||||
let unknownCount = 0;
|
||||
let wrongContentCount = 0;
|
||||
|
||||
// Batch size for getMessages. TDLib accepts up to ~100 IDs per call.
|
||||
// Using 100 means 20k packages → ~200 round-trips instead of 20k.
|
||||
const BATCH_SIZE = 100;
|
||||
|
||||
for (const [, channelPackages] of byChannel) {
|
||||
for (const pkg of channelPackages) {
|
||||
const exists = await verifyMessageExists(
|
||||
// Group packages by destChannelId (already done) — within each group,
|
||||
// process in batches via getMessages (plural).
|
||||
for (let i = 0; i < channelPackages.length; i += BATCH_SIZE) {
|
||||
const batch = channelPackages.slice(i, i + BATCH_SIZE);
|
||||
const batchResults = await verifyMessagesBatch(
|
||||
client,
|
||||
destChannel.telegramId,
|
||||
pkg.destMessageId!
|
||||
batch.map((p) => p.destMessageId!)
|
||||
);
|
||||
|
||||
if (exists) {
|
||||
verifiedCount++;
|
||||
} else {
|
||||
for (let j = 0; j < batch.length; j++) {
|
||||
const pkg = batch[j];
|
||||
const result = batchResults[j];
|
||||
|
||||
if (result.state === "exists") {
|
||||
verifiedCount++;
|
||||
} else if (result.state === "deleted") {
|
||||
log.warn(
|
||||
{
|
||||
packageId: pkg.id,
|
||||
@@ -100,21 +113,50 @@ export async function recoverIncompleteUploads(): Promise<void> {
|
||||
);
|
||||
await resetPackageDestination(pkg.id);
|
||||
resetCount++;
|
||||
} else if (result.state === "wrong-content") {
|
||||
// The message exists but isn't a document anymore (got cleared /
|
||||
// replaced). Treat as missing so we re-upload.
|
||||
log.warn(
|
||||
{
|
||||
packageId: pkg.id,
|
||||
fileName: pkg.fileName,
|
||||
destMessageId: Number(pkg.destMessageId),
|
||||
contentType: result.contentType,
|
||||
},
|
||||
"Destination message is not a document, resetting package for re-upload"
|
||||
);
|
||||
await resetPackageDestination(pkg.id);
|
||||
wrongContentCount++;
|
||||
} else {
|
||||
// Unknown — TDLib couldn't tell us. Don't reset, but DO count this
|
||||
// so the summary line shows recovery wasn't 100% successful.
|
||||
unknownCount++;
|
||||
log.warn(
|
||||
{
|
||||
packageId: pkg.id,
|
||||
fileName: pkg.fileName,
|
||||
destMessageId: Number(pkg.destMessageId),
|
||||
reason: result.reason.slice(0, 200),
|
||||
},
|
||||
"Could not verify destination message — will retry on next startup"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (resetCount > 0) {
|
||||
log.info(
|
||||
{ resetCount, verifiedCount, totalChecked: packages.length },
|
||||
"Upload recovery complete — packages reset for re-processing"
|
||||
);
|
||||
} else {
|
||||
log.info(
|
||||
{ verifiedCount, totalChecked: packages.length },
|
||||
"Upload recovery complete — all destination messages verified"
|
||||
);
|
||||
}
|
||||
log.info(
|
||||
{
|
||||
verifiedCount,
|
||||
resetCount,
|
||||
wrongContentCount,
|
||||
unknownCount,
|
||||
totalChecked: packages.length,
|
||||
},
|
||||
unknownCount === 0
|
||||
? "Upload recovery complete"
|
||||
: "Upload recovery complete — some packages could not be verified, will retry next startup"
|
||||
);
|
||||
} catch (err) {
|
||||
log.error({ err }, "Upload recovery failed (non-fatal, will retry next startup)");
|
||||
} finally {
|
||||
@@ -124,15 +166,77 @@ export async function recoverIncompleteUploads(): Promise<void> {
|
||||
}
|
||||
}
|
||||
|
||||
type VerifyResult =
|
||||
| { state: "exists" }
|
||||
| { state: "deleted" }
|
||||
| { state: "wrong-content"; contentType: string }
|
||||
| { state: "unknown"; reason: string };
|
||||
|
||||
/**
|
||||
* Check whether a message exists in a Telegram chat.
|
||||
* Returns false if the message was deleted or never existed.
|
||||
* Batch version of verifyMessageExists. Calls TDLib's getMessages (plural)
|
||||
* with up to ~100 message IDs at once. Returns one VerifyResult per input
|
||||
* ID, in input order. Missing messages come back as null in TDLib's response
|
||||
* — translated to {state: "deleted"} here.
|
||||
*
|
||||
* Falls back to per-message verification on any error so that one bad batch
|
||||
* doesn't lose all verification for that chunk.
|
||||
*/
|
||||
async function verifyMessagesBatch(
|
||||
client: Client,
|
||||
chatTelegramId: bigint,
|
||||
messageIds: bigint[]
|
||||
): Promise<VerifyResult[]> {
|
||||
try {
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const result = (await withFloodWait(
|
||||
() =>
|
||||
client.invoke({
|
||||
_: "getMessages",
|
||||
chat_id: Number(chatTelegramId),
|
||||
message_ids: messageIds.map((id) => Number(id)),
|
||||
}),
|
||||
"getMessages:verify"
|
||||
)) as { messages?: (null | { content?: { _: string } })[] };
|
||||
|
||||
const messages = result.messages ?? [];
|
||||
return messageIds.map((_id, i) => {
|
||||
const m = messages[i];
|
||||
if (!m || !m.content) return { state: "deleted" };
|
||||
if (m.content._ !== "messageDocument") {
|
||||
return { state: "wrong-content", contentType: String(m.content._) };
|
||||
}
|
||||
return { state: "exists" };
|
||||
});
|
||||
} catch (err) {
|
||||
// If the whole batch errors out, fall back to per-message verification.
|
||||
log.warn(
|
||||
{ err, batchSize: messageIds.length, chatTelegramId: chatTelegramId.toString() },
|
||||
"getMessages batch failed, falling back to per-message verification"
|
||||
);
|
||||
const out: VerifyResult[] = [];
|
||||
for (const id of messageIds) {
|
||||
out.push(await verifyMessageExists(client, chatTelegramId, id));
|
||||
}
|
||||
return out;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check whether a message exists in a Telegram chat and is the document we
|
||||
* uploaded. Returns a discriminated result instead of a bare boolean so the
|
||||
* caller can distinguish "definitely gone" (reset) from "couldn't reach TG"
|
||||
* (leave alone, try again next startup).
|
||||
*
|
||||
* Previous version conflated all non-404 errors with "exists", which masked
|
||||
* recovery completely when TDLib had a degraded connection — the worker
|
||||
* would log "all destination messages verified" even though it had answered
|
||||
* questions it couldn't actually answer.
|
||||
*/
|
||||
async function verifyMessageExists(
|
||||
client: Client,
|
||||
chatTelegramId: bigint,
|
||||
messageId: bigint
|
||||
): Promise<boolean> {
|
||||
): Promise<VerifyResult> {
|
||||
try {
|
||||
const result = await withFloodWait(
|
||||
() =>
|
||||
@@ -144,44 +248,37 @@ async function verifyMessageExists(
|
||||
"getMessage:verify"
|
||||
);
|
||||
|
||||
// TDLib returns the message object if it exists.
|
||||
// A deleted message may return with content type "messageChatDeleteMessage"
|
||||
// or the call may throw. Check that we got a real message with content.
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const msg = result as any;
|
||||
if (!msg || !msg.content) {
|
||||
return false;
|
||||
return { state: "deleted" };
|
||||
}
|
||||
|
||||
// Check that the message has document content (our uploads are documents)
|
||||
// A message that exists but has no document content was likely cleared/replaced
|
||||
if (msg.content._ !== "messageDocument") {
|
||||
log.debug(
|
||||
{
|
||||
messageId: Number(messageId),
|
||||
contentType: msg.content._,
|
||||
},
|
||||
"Destination message exists but is not a document"
|
||||
);
|
||||
return false;
|
||||
return { state: "wrong-content", contentType: String(msg.content._) };
|
||||
}
|
||||
|
||||
return true;
|
||||
return { state: "exists" };
|
||||
} catch (err) {
|
||||
// TDLib throws "Message not found" (error code 404) for deleted messages
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
const errMessage = err instanceof Error ? err.message : String(err);
|
||||
const code = (err as { code?: number })?.code;
|
||||
|
||||
if (code === 404 || message.includes("not found") || message.includes("Not Found")) {
|
||||
return false;
|
||||
// Hard "the message is definitely gone" signals from TDLib:
|
||||
// - HTTP 404
|
||||
// - "Message not found" / "MESSAGE_ID_INVALID" error strings
|
||||
const lower = errMessage.toLowerCase();
|
||||
if (
|
||||
code === 404 ||
|
||||
lower.includes("message not found") ||
|
||||
lower.includes("message_id_invalid") ||
|
||||
lower.includes("messageidinvalid") ||
|
||||
lower.includes("not found")
|
||||
) {
|
||||
return { state: "deleted" };
|
||||
}
|
||||
|
||||
// For other errors (network issues, etc.), assume the message exists
|
||||
// to avoid incorrectly resetting packages due to transient failures
|
||||
log.warn(
|
||||
{ err, messageId: Number(messageId) },
|
||||
"Could not verify message (assuming it exists)"
|
||||
);
|
||||
return true;
|
||||
// Everything else (network, connection, TDLib internal) is genuinely
|
||||
// unknown — do NOT claim "verified".
|
||||
return { state: "unknown", reason: errMessage };
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,6 +5,36 @@ import { withFloodWait } from "../util/retry.js";
|
||||
|
||||
const log = childLogger("chats");
|
||||
|
||||
/**
|
||||
* Collect chat folder IDs to widen the loadChats sweep across all folder
|
||||
* chat lists. In TDLib 1.8.64+ there's no synchronous getChatFolders call —
|
||||
* the folder list arrives via updateChatFolders. We listen for it briefly
|
||||
* (200ms) and fall back to an empty list if nothing arrives; chats inside
|
||||
* folders are still reachable via chatListMain so this only loses some
|
||||
* preemptive cache warming.
|
||||
*/
|
||||
async function collectFolderIds(
|
||||
client: Client
|
||||
): Promise<{ _: "chatListFolder"; chat_folder_id: number }[]> {
|
||||
return new Promise((resolve) => {
|
||||
const ids: number[] = [];
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const handler = (update: any) => {
|
||||
if (update?._ === "updateChatFolders") {
|
||||
const folders = update.chat_folders as { id: number }[] | undefined;
|
||||
if (folders) {
|
||||
for (const f of folders) ids.push(f.id);
|
||||
}
|
||||
}
|
||||
};
|
||||
client.on("update", handler);
|
||||
setTimeout(() => {
|
||||
client.off("update", handler);
|
||||
resolve(ids.map((id) => ({ _: "chatListFolder" as const, chat_folder_id: id })));
|
||||
}, 200);
|
||||
});
|
||||
}
|
||||
|
||||
export interface TelegramChatInfo {
|
||||
chatId: bigint;
|
||||
title: string;
|
||||
@@ -37,21 +67,16 @@ export async function getAccountChats(
|
||||
// First, load all chats into TDLib's cache using loadChats (the proper API).
|
||||
// loadChats returns 404 when all chats have been loaded.
|
||||
// Then use getChats to retrieve the IDs for enrichment.
|
||||
// Load from main, archive, AND chat folders to cover all chat types.
|
||||
const folderLists: { _: "chatListFolder"; chat_folder_id: number }[] = [];
|
||||
try {
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const folders = (await client.invoke({ _: "getChatFolders" })) as any;
|
||||
if (folders?.chat_folders) {
|
||||
for (const f of folders.chat_folders) {
|
||||
folderLists.push({ _: "chatListFolder", chat_folder_id: f.id });
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// getChatFolders may not be available in older TDLib versions
|
||||
}
|
||||
//
|
||||
// Folder-specific loading (chatListFolder) was removed in TDLib 1.8.64+ —
|
||||
// getChatFolders (plural) is no longer a callable method, only the
|
||||
// updateChatFolders event. The chats inside folders are still reachable
|
||||
// via chatListMain so this isn't a functional regression.
|
||||
const folderLists: { _: "chatListFolder"; chat_folder_id: number }[] =
|
||||
await collectFolderIds(client);
|
||||
|
||||
const chatLists: Record<string, unknown>[] = [
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const chatLists: any[] = [
|
||||
{ _: "chatListMain" },
|
||||
{ _: "chatListArchive" },
|
||||
...folderLists,
|
||||
|
||||
@@ -51,6 +51,10 @@ interface TdMessage {
|
||||
path?: string;
|
||||
is_downloading_completed?: boolean;
|
||||
};
|
||||
remote?: {
|
||||
/** Stable identifier across reposts of the same file content. */
|
||||
unique_id?: string;
|
||||
};
|
||||
};
|
||||
};
|
||||
photo?: {
|
||||
@@ -115,7 +119,10 @@ export async function invokeWithTimeout<T>(
|
||||
}
|
||||
}, timeoutMs);
|
||||
|
||||
(client.invoke(request) as Promise<T>)
|
||||
// The tdl 8.1+ types are very strict about the literal `_` field;
|
||||
// our generic wrapper passes arbitrary requests, so cast through any.
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
(client.invoke(request as any) as Promise<T>)
|
||||
.then((result) => {
|
||||
if (!settled) {
|
||||
settled = true;
|
||||
@@ -228,6 +235,7 @@ export async function getChannelMessages(
|
||||
mediaAlbumId: msg.media_album_id && msg.media_album_id !== "0" ? msg.media_album_id : undefined,
|
||||
replyToMessageId: msg.reply_to_message_id ? BigInt(msg.reply_to_message_id) : undefined,
|
||||
caption: msg.content?.caption?.text || undefined,
|
||||
remoteUniqueId: doc.document.remote?.unique_id || undefined,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -210,6 +210,7 @@ export async function getTopicMessages(
|
||||
document?: {
|
||||
id: number;
|
||||
size: number;
|
||||
remote?: { unique_id?: string };
|
||||
};
|
||||
};
|
||||
photo?: {
|
||||
@@ -257,6 +258,7 @@ export async function getTopicMessages(
|
||||
fileSize: BigInt(doc.document.size),
|
||||
date: new Date(msg.date * 1000),
|
||||
mediaAlbumId: msg.media_album_id && msg.media_album_id !== "0" ? msg.media_album_id : undefined,
|
||||
remoteUniqueId: doc.document.remote?.unique_id || undefined,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -31,6 +31,10 @@ import {
|
||||
upsertSkippedPackage,
|
||||
deleteSkippedPackage,
|
||||
getCappedSkippedMessageIds,
|
||||
findRepostedPackage,
|
||||
findPackageByRemoteUniqueId,
|
||||
getRetryableSkippedMessageIds,
|
||||
updatePackageTopicContext,
|
||||
} from "./db/queries.js";
|
||||
import type { ActivityUpdate } from "./db/queries.js";
|
||||
import { createTdlibClient, closeTdlibClient } from "./tdlib/client.js";
|
||||
@@ -43,6 +47,8 @@ import { pickPreviewFile, extractPreviewImage } from "./preview/extract.js";
|
||||
import { groupArchiveSets } from "./archive/multipart.js";
|
||||
import type { ArchiveSet } from "./archive/multipart.js";
|
||||
import { extractCreatorFromFileName, extractCreatorFromChannelTitle } from "./archive/creator.js";
|
||||
import { extractSlicerTags } from "./archive/slicer-tags.js";
|
||||
import { testArchiveIntegrity } from "./archive/integrity.js";
|
||||
import { hashParts } from "./archive/hash.js";
|
||||
import { readZipCentralDirectory } from "./archive/zip-reader.js";
|
||||
import { readRarContents } from "./archive/rar-reader.js";
|
||||
@@ -354,24 +360,32 @@ export async function runWorkerForAccount(
|
||||
// Load all chats into TDLib's local cache using loadChats (the recommended API).
|
||||
// Without this, getChat/searchChatMessages fail with "Chat not found".
|
||||
// loadChats returns a 404 when all chats have been loaded — that's the stop signal.
|
||||
// Load from main, archive, AND chat folders to cover all chat types.
|
||||
//
|
||||
// TDLib 1.8.64+ removed the synchronous getChatFolders call; folder IDs
|
||||
// now arrive only via the updateChatFolders event. We listen briefly,
|
||||
// then load main + archive + any folders we caught. Chats inside folders
|
||||
// are also reachable from chatListMain so missing the folder sweep is
|
||||
// not a functional regression — it just loses a small bit of cache warming.
|
||||
{
|
||||
// Discover chat folders first
|
||||
const folderLists: { _: "chatListFolder"; chat_folder_id: number }[] = [];
|
||||
try {
|
||||
const folders = await client.invoke({ _: "getChatFolders" }) as {
|
||||
chat_folders?: { id: number }[];
|
||||
};
|
||||
if (folders.chat_folders) {
|
||||
for (const f of folders.chat_folders) {
|
||||
folderLists.push({ _: "chatListFolder", chat_folder_id: f.id });
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const folderLists: any[] = await new Promise((resolve) => {
|
||||
const ids: number[] = [];
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const handler = (update: any) => {
|
||||
if (update?._ === "updateChatFolders") {
|
||||
const folders = update.chat_folders as { id: number }[] | undefined;
|
||||
if (folders) for (const f of folders) ids.push(f.id);
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// getChatFolders may not be available in older TDLib versions
|
||||
}
|
||||
};
|
||||
client.on("update", handler);
|
||||
setTimeout(() => {
|
||||
client.off("update", handler);
|
||||
resolve(ids.map((id) => ({ _: "chatListFolder", chat_folder_id: id })));
|
||||
}, 200);
|
||||
});
|
||||
|
||||
const chatLists: Record<string, unknown>[] = [
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const chatLists: any[] = [
|
||||
{ _: "chatListMain" },
|
||||
{ _: "chatListArchive" },
|
||||
...folderLists,
|
||||
@@ -523,21 +537,77 @@ export async function runWorkerForAccount(
|
||||
messagesScanned: counters.messagesScanned,
|
||||
});
|
||||
|
||||
const topics = await getForumTopicList(client, channel.telegramId);
|
||||
const rawTopics = await getForumTopicList(client, channel.telegramId);
|
||||
const topicProgressList = await getTopicProgress(mapping.id);
|
||||
|
||||
// Process more-specific topics BEFORE "General" so the first
|
||||
// encounter of any file is in its most specific context. This makes
|
||||
// newly-created Packages carry useful topic info (e.g., a campaign
|
||||
// name) instead of just "General".
|
||||
const topics = [...rawTopics].sort((a, b) => {
|
||||
const aIsGeneral = a.name === "General";
|
||||
const bIsGeneral = b.name === "General";
|
||||
if (aIsGeneral === bIsGeneral) return 0;
|
||||
return aIsGeneral ? 1 : -1;
|
||||
});
|
||||
|
||||
accountLog.info(
|
||||
{ channelId: channel.id, title: channel.title, topicCount: topics.length },
|
||||
"Scanning forum channel by topic"
|
||||
"Scanning forum channel by topic (specific topics first, General last)"
|
||||
);
|
||||
|
||||
for (let tIdx = 0; tIdx < topics.length; tIdx++) {
|
||||
const topic = topics[tIdx];
|
||||
try {
|
||||
const progress = topicProgressList.find(
|
||||
let progress = topicProgressList.find(
|
||||
(tp) => tp.topicId === topic.topicId
|
||||
);
|
||||
|
||||
// ── SkippedPackage retry pass ──
|
||||
// If we have failed messages in this topic with attemptCount
|
||||
// below the cap, pull the watermark back below the lowest of
|
||||
// them so the scan re-picks them up. Without this, a message
|
||||
// that failed before my watermark cap fix (or had its watermark
|
||||
// advanced past it via the all-failures fallback) is stuck in
|
||||
// SkippedPackage forever.
|
||||
try {
|
||||
const retryable = await getRetryableSkippedMessageIds({
|
||||
accountId: account.id,
|
||||
sourceChannelId: channel.id,
|
||||
topicId: topic.topicId,
|
||||
cap: config.maxSkipAttempts,
|
||||
});
|
||||
if (retryable.length > 0) {
|
||||
const lowest = retryable[0];
|
||||
const currentWatermark = progress?.lastProcessedMessageId ?? null;
|
||||
if (currentWatermark !== null && currentWatermark >= lowest) {
|
||||
const resetTo = lowest - 1n;
|
||||
await upsertTopicProgress(
|
||||
mapping.id,
|
||||
topic.topicId,
|
||||
topic.name,
|
||||
resetTo
|
||||
);
|
||||
accountLog.info(
|
||||
{
|
||||
topic: topic.name,
|
||||
retryableCount: retryable.length,
|
||||
lowestSkippedMsgId: lowest.toString(),
|
||||
oldWatermark: currentWatermark.toString(),
|
||||
newWatermark: resetTo.toString(),
|
||||
},
|
||||
"Resetting topic watermark to retry skipped messages"
|
||||
);
|
||||
progress = { ...(progress ?? { id: "", accountChannelMapId: mapping.id, topicId: topic.topicId, topicName: topic.name }), lastProcessedMessageId: resetTo } as typeof progress;
|
||||
}
|
||||
}
|
||||
} catch (retryErr) {
|
||||
accountLog.warn(
|
||||
{ err: retryErr, topic: topic.name },
|
||||
"SkippedPackage retry pass failed (non-fatal)"
|
||||
);
|
||||
}
|
||||
|
||||
const topicLabel = `${channel.title} › ${topic.name}`;
|
||||
const topicProgress = topics.length > 1
|
||||
? ` (topic ${tIdx + 1}/${topics.length})`
|
||||
@@ -663,10 +733,47 @@ export async function runWorkerForAccount(
|
||||
"Processing source channel"
|
||||
);
|
||||
|
||||
// ── SkippedPackage retry pass ──
|
||||
// Pull the watermark back below the lowest still-retryable
|
||||
// SkippedPackage so they get picked up by the scan. See the matching
|
||||
// block in the forum branch for the rationale.
|
||||
let effectiveChannelWatermark = mapping.lastProcessedMessageId;
|
||||
try {
|
||||
const retryable = await getRetryableSkippedMessageIds({
|
||||
accountId: account.id,
|
||||
sourceChannelId: channel.id,
|
||||
topicId: null,
|
||||
cap: config.maxSkipAttempts,
|
||||
});
|
||||
if (retryable.length > 0) {
|
||||
const lowest = retryable[0];
|
||||
if (effectiveChannelWatermark !== null && effectiveChannelWatermark >= lowest) {
|
||||
const resetTo = lowest - 1n;
|
||||
await updateLastProcessedMessage(mapping.id, resetTo);
|
||||
accountLog.info(
|
||||
{
|
||||
channel: channel.title,
|
||||
retryableCount: retryable.length,
|
||||
lowestSkippedMsgId: lowest.toString(),
|
||||
oldWatermark: effectiveChannelWatermark.toString(),
|
||||
newWatermark: resetTo.toString(),
|
||||
},
|
||||
"Resetting channel watermark to retry skipped messages"
|
||||
);
|
||||
effectiveChannelWatermark = resetTo;
|
||||
}
|
||||
}
|
||||
} catch (retryErr) {
|
||||
accountLog.warn(
|
||||
{ err: retryErr, channel: channel.title },
|
||||
"SkippedPackage retry pass failed (non-fatal)"
|
||||
);
|
||||
}
|
||||
|
||||
const scanResult = await getChannelMessages(
|
||||
client,
|
||||
channel.telegramId,
|
||||
mapping.lastProcessedMessageId,
|
||||
effectiveChannelWatermark,
|
||||
100,
|
||||
(scanned) => {
|
||||
throttled.update({
|
||||
@@ -705,7 +812,7 @@ export async function runWorkerForAccount(
|
||||
pipelineCtx,
|
||||
scanResult,
|
||||
run.id,
|
||||
mapping.lastProcessedMessageId,
|
||||
effectiveChannelWatermark,
|
||||
// Incremental watermark advance — saves progress per-set so a
|
||||
// worker restart mid-scan doesn't lose all work.
|
||||
async (messageId) => {
|
||||
@@ -1136,6 +1243,38 @@ async function processOneArchiveSet(
|
||||
|
||||
const archiveName = archiveSet.parts[0].fileName;
|
||||
|
||||
// ── Earliest skip: remote.unique_id match ──
|
||||
// TDLib reports a stable unique_id per file content. If we already have a
|
||||
// Package in this channel with the same unique_id, it's the exact same
|
||||
// file content reposted at a new message ID — zero false positives.
|
||||
const firstRemoteUniqueId = archiveSet.parts[0].remoteUniqueId;
|
||||
if (firstRemoteUniqueId) {
|
||||
const match = await findPackageByRemoteUniqueId(channel.id, firstRemoteUniqueId);
|
||||
if (match) {
|
||||
counters.zipsDuplicate++;
|
||||
accountLog.info(
|
||||
{
|
||||
fileName: archiveSet.parts[0].fileName,
|
||||
sourceMessageId: Number(archiveSet.parts[0].id),
|
||||
remoteUniqueId: firstRemoteUniqueId,
|
||||
existingPackageId: match.id,
|
||||
existingDestMessageId: match.destMessageId ? Number(match.destMessageId) : null,
|
||||
},
|
||||
"Skipping — remote.unique_id matches an existing Package in this channel"
|
||||
);
|
||||
await updateRunActivity(runId, {
|
||||
currentActivity: `Skipped ${archiveSet.parts[0].fileName} (already ingested by unique_id)`,
|
||||
currentStep: "deduplicating",
|
||||
currentChannel: channelTitle,
|
||||
currentFile: archiveSet.parts[0].fileName,
|
||||
currentFileNum: setIdx + 1,
|
||||
totalFiles: totalSets,
|
||||
zipsDuplicate: counters.zipsDuplicate,
|
||||
});
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
// ── Early skip: check if this archive set was already ingested ──
|
||||
// This avoids re-downloading large archives that were processed in a prior run.
|
||||
const alreadyIngested = await packageExistsBySourceMessage(
|
||||
@@ -1160,8 +1299,83 @@ async function processOneArchiveSet(
|
||||
return null;
|
||||
}
|
||||
|
||||
// ── Size guard: skip archives that exceed WORKER_MAX_ZIP_SIZE_MB ──
|
||||
// Compute the total size across all parts (used by the repost check below
|
||||
// AND by the size guard further down).
|
||||
const totalArchiveSize = archiveSet.parts.reduce((sum, p) => sum + p.fileSize, 0n);
|
||||
|
||||
// ── Pre-download repost detection ──
|
||||
// The source channel admin frequently reposts the same file at new message
|
||||
// IDs. packageExistsBySourceMessage misses these (different msgId), so we
|
||||
// historically downloaded the file just to discover via hash that it's a
|
||||
// duplicate — wasting hours of bandwidth per run.
|
||||
//
|
||||
// Match by (sourceChannelId, fileName, totalSize). The totalSize comparison
|
||||
// makes this very strong — name-and-size collision between unrelated files
|
||||
// is rare in practice. If it ever happens, the new file is treated as a
|
||||
// duplicate; the user can remove the existing Package via the UI to force
|
||||
// a re-ingestion.
|
||||
const reposted = await findRepostedPackage(
|
||||
channel.id,
|
||||
archiveName,
|
||||
totalArchiveSize
|
||||
);
|
||||
if (reposted) {
|
||||
counters.zipsDuplicate++;
|
||||
|
||||
// Backfill topic context onto the existing Package when we encounter the
|
||||
// same file in a more specific topic. If the existing Package was created
|
||||
// from "General" or a non-forum scan and we now see the file in a named
|
||||
// topic (e.g., "Artisan Guild January 2022"), update the Package so the
|
||||
// user gets richer metadata. We only update when the current scan is in
|
||||
// a specific topic AND the existing topic differs.
|
||||
const currentTopicName = ctx.topicCreator; // == topic.name for forum scans
|
||||
const currentTopicId = ctx.sourceTopicId;
|
||||
const isCurrentSpecific = currentTopicName !== null && currentTopicName !== "General";
|
||||
const existingTopicDiffers = reposted.sourceTopicId !== currentTopicId;
|
||||
if (isCurrentSpecific && currentTopicId !== null && existingTopicDiffers) {
|
||||
try {
|
||||
await updatePackageTopicContext(reposted.id, currentTopicId, currentTopicName);
|
||||
accountLog.info(
|
||||
{
|
||||
fileName: archiveName,
|
||||
packageId: reposted.id,
|
||||
existingTopicId: reposted.sourceTopicId ? Number(reposted.sourceTopicId) : null,
|
||||
newTopicId: Number(currentTopicId),
|
||||
newTopicName: currentTopicName,
|
||||
},
|
||||
"Updated existing Package with more specific topic context"
|
||||
);
|
||||
} catch (updErr) {
|
||||
accountLog.warn({ err: updErr, packageId: reposted.id }, "Failed to update Package topic context (non-fatal)");
|
||||
}
|
||||
}
|
||||
|
||||
accountLog.info(
|
||||
{
|
||||
fileName: archiveName,
|
||||
sourceMessageId: Number(archiveSet.parts[0].id),
|
||||
existingPackageId: reposted.id,
|
||||
existingDestMessageId: reposted.destMessageId ? Number(reposted.destMessageId) : null,
|
||||
existingTopicId: reposted.sourceTopicId ? Number(reposted.sourceTopicId) : null,
|
||||
currentTopicId: currentTopicId ? Number(currentTopicId) : null,
|
||||
currentTopicName,
|
||||
totalSize: Number(totalArchiveSize),
|
||||
},
|
||||
"Skipping repost — same fileName + size already uploaded in this channel"
|
||||
);
|
||||
await updateRunActivity(runId, {
|
||||
currentActivity: `Skipped ${archiveName} (repost of already-uploaded file)`,
|
||||
currentStep: "deduplicating",
|
||||
currentChannel: channelTitle,
|
||||
currentFile: archiveName,
|
||||
currentFileNum: setIdx + 1,
|
||||
totalFiles: totalSets,
|
||||
zipsDuplicate: counters.zipsDuplicate,
|
||||
});
|
||||
return null;
|
||||
}
|
||||
|
||||
// ── Size guard: skip archives that exceed WORKER_MAX_ZIP_SIZE_MB ──
|
||||
const maxSizeBytes = BigInt(config.maxZipSizeMB) * 1024n * 1024n;
|
||||
if (totalArchiveSize > maxSizeBytes) {
|
||||
accountLog.warn(
|
||||
@@ -1440,6 +1654,19 @@ async function processOneArchiveSet(
|
||||
);
|
||||
}
|
||||
|
||||
// ── Pre-upload integrity test ──
|
||||
// Catch broken/encrypted archives before we burn upload bandwidth on
|
||||
// them. Cheap (unzip -t / unrar t / 7z t) compared to a multi-GB upload.
|
||||
// Skipped when we're reusing an existing upload — no point testing the
|
||||
// file again.
|
||||
const integrity = await testArchiveIntegrity(
|
||||
archiveSet.type === "7Z" ? "SEVEN_Z" : archiveSet.type,
|
||||
uploadPaths[0]
|
||||
);
|
||||
if (!integrity.ok) {
|
||||
throw new Error(`Archive integrity check failed: ${integrity.reason}`);
|
||||
}
|
||||
|
||||
// ── Uploading ──
|
||||
// Check if a prior run already uploaded this file (orphaned upload scenario:
|
||||
// file reached Telegram but DB write failed or worker crashed before indexing)
|
||||
@@ -1505,6 +1732,72 @@ async function processOneArchiveSet(
|
||||
}
|
||||
}
|
||||
|
||||
// ── Destination read-back verification ──
|
||||
// Telegram's updateMessageSendSucceeded fires when TG acknowledges the
|
||||
// message, but that's separate from "the message is queryable and
|
||||
// contains the file we sent". Fetch each destination message and
|
||||
// confirm the document's size matches what we uploaded.
|
||||
//
|
||||
// Skipped when reusing an existing upload (we never sent anything).
|
||||
// Failures here surface as a SystemNotification but DO NOT abort the
|
||||
// ingestion — the Package will be created with whatever destMessageIds
|
||||
// Telegram returned, and a future recovery run can reset it if needed.
|
||||
if (!existingUpload && destResult.messageIds.length > 0) {
|
||||
try {
|
||||
const expectedSizes = uploadPaths.length === destResult.messageIds.length
|
||||
? await Promise.all(
|
||||
uploadPaths.map(async (p) => (await import("fs/promises")).stat(p).then((s) => s.size))
|
||||
)
|
||||
: null;
|
||||
|
||||
for (let i = 0; i < destResult.messageIds.length; i++) {
|
||||
const msgId = Number(destResult.messageIds[i]);
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
const tdMsg = (await client.invoke({
|
||||
_: "getMessage",
|
||||
chat_id: Number(destChannelTelegramId),
|
||||
message_id: msgId,
|
||||
}).catch(() => null)) as any;
|
||||
|
||||
const doc = tdMsg?.content?.document?.document;
|
||||
const actualSize = doc?.size;
|
||||
const expected = expectedSizes?.[i];
|
||||
|
||||
if (!actualSize) {
|
||||
accountLog.warn(
|
||||
{ fileName: archiveName, destMessageId: msgId },
|
||||
"Post-upload read-back: destination message has no document content"
|
||||
);
|
||||
await db.systemNotification.create({
|
||||
data: {
|
||||
type: "UPLOAD_FAILED",
|
||||
severity: "WARNING",
|
||||
title: `Read-back failed: ${archiveName}`,
|
||||
message: `Destination message ${msgId} has no document content after upload. The upload may have failed silently.`,
|
||||
context: { fileName: archiveName, destMessageId: msgId, sourceChannelId: channel.id },
|
||||
},
|
||||
});
|
||||
} else if (expected !== undefined && actualSize !== expected) {
|
||||
accountLog.error(
|
||||
{ fileName: archiveName, destMessageId: msgId, expectedSize: expected, actualSize },
|
||||
"Post-upload read-back: destination file size mismatch"
|
||||
);
|
||||
await db.systemNotification.create({
|
||||
data: {
|
||||
type: "HASH_MISMATCH",
|
||||
severity: "ERROR",
|
||||
title: `Read-back size mismatch: ${archiveName}`,
|
||||
message: `Sent ${expected} bytes but destination message ${msgId} contains a ${actualSize}-byte file.`,
|
||||
context: { fileName: archiveName, destMessageId: msgId, expectedSize: expected, actualSize, sourceChannelId: channel.id },
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
} catch (readBackErr) {
|
||||
accountLog.warn({ err: readBackErr, fileName: archiveName }, "Post-upload read-back failed (non-fatal)");
|
||||
}
|
||||
}
|
||||
|
||||
// ── Phase 1: Stub record — persisted immediately after upload ──
|
||||
await deleteOrphanedPackageByHash(contentHash);
|
||||
|
||||
@@ -1518,6 +1811,15 @@ async function processOneArchiveSet(
|
||||
tags.push(channel.category);
|
||||
}
|
||||
|
||||
// Derive slicer tags from the file listing so users can filter the
|
||||
// catalog by "what software opens these files". Tags include "lychee",
|
||||
// "chitubox", "anycubic", "bambu", "fdm" etc. — only added if matching
|
||||
// slicer-specific files are present in the archive.
|
||||
const slicerTags = extractSlicerTags(entries);
|
||||
for (const tag of slicerTags) {
|
||||
if (!tags.includes(tag)) tags.push(tag);
|
||||
}
|
||||
|
||||
stub = await createPackageStub({
|
||||
contentHash,
|
||||
fileName: archiveName,
|
||||
@@ -1526,6 +1828,7 @@ async function processOneArchiveSet(
|
||||
sourceChannelId: channel.id,
|
||||
sourceMessageId: archiveSet.parts[0].id,
|
||||
sourceTopicId,
|
||||
remoteUniqueId: archiveSet.parts[0].remoteUniqueId ?? null,
|
||||
destChannelId,
|
||||
destMessageId: destResult.messageId,
|
||||
destMessageIds: destResult.messageIds,
|
||||
|
||||
Reference in New Issue
Block a user