(sourceId: string, ds: DataSource, sess: ImportSession)
| 216 | } |
| 217 | |
| 218 | async executePullSession(sourceId: string, ds: DataSource, sess: ImportSession): Promise<PullSessionResult> { |
| 219 | const currentDs = this.dsManager.get(sourceId) |
| 220 | if (!currentDs || !currentDs.sessions.some((s) => s.id === sess.id)) { |
| 221 | this.logger.info(`[Pull] Skipping "${sess.name}": session no longer exists`) |
| 222 | return { success: true, newMessageCount: 0 } |
| 223 | } |
| 224 | |
| 225 | const importStatusSessionId = sess.targetSessionId || deriveLocalSessionId(ds.baseUrl, sess.remoteSessionId) |
| 226 | if (this.isImporting(importStatusSessionId)) { |
| 227 | this.logger.info(`[Pull] Skipping "${sess.name}": import in progress`) |
| 228 | return { success: false, newMessageCount: 0, error: 'Import in progress' } |
| 229 | } |
| 230 | |
| 231 | this.logger.info(`[Pull] Pulling "${sess.name}" from ${ds.baseUrl}`) |
| 232 | |
| 233 | let totalNewMessages = 0 |
| 234 | let since = sess.lastPullAt |
| 235 | let offset: number | undefined |
| 236 | let nextPullSince = sess.lastPullAt |
| 237 | let pageCount = 0 |
| 238 | let resyncAttempted = false |
| 239 | |
| 240 | this.progressMap.set(sess.id, { sessionId: sess.id, sessionName: sess.name, current: 0, pages: 0, done: false }) |
| 241 | |
| 242 | try { |
| 243 | while (pageCount < MAX_PAGES_PER_PULL) { |
| 244 | pageCount++ |
| 245 | const tempFile = await this.fetcher.fetchToTempFile(ds.baseUrl, sess.remoteSessionId, ds.token, { |
| 246 | since, |
| 247 | offset, |
| 248 | limit: ds.pullLimit, |
| 249 | }) |
| 250 | |
| 251 | try { |
| 252 | const stat = fs.statSync(tempFile) |
| 253 | this.logger.info(`[Pull] "${sess.name}" page ${pageCount}: fetched ${stat.size} bytes`) |
| 254 | |
| 255 | const sync0 = parseSyncFromFile(tempFile) |
| 256 | if (stat.size < 1024) { |
| 257 | if (!fileContainsMessages(tempFile) && sync0?.hasMore === false) { |
| 258 | cleanupTempFile(tempFile) |
| 259 | if (sync0?.nextSince !== undefined) nextPullSince = Math.max(nextPullSince, sync0.nextSince) |
| 260 | const retryDelays = [2000, 3000, 5000] |
| 261 | let retrySuccess = false |
| 262 | let retryHasMore = false |
| 263 | for (let ri = 0; ri < retryDelays.length; ri++) { |
| 264 | this.logger.info( |
| 265 | `[Pull] "${sess.name}" page ${pageCount} got empty response, retry ${ri + 1}/${retryDelays.length} after ${retryDelays[ri]}ms` |
| 266 | ) |
| 267 | await new Promise((r) => setTimeout(r, retryDelays[ri])) |
| 268 | const retryFile = await this.fetcher.fetchToTempFile(ds.baseUrl, sess.remoteSessionId, ds.token, { |
| 269 | since, |
| 270 | offset, |
| 271 | limit: ds.pullLimit, |
| 272 | }) |
| 273 | const retryStat = fs.statSync(retryFile) |
| 274 | this.logger.info(`[Pull] "${sess.name}" retry ${ri + 1}: fetched ${retryStat.size} bytes`) |
| 275 | const retrySync = parseSyncFromFile(retryFile) |
no test coverage detected