From 189e0ad227da9b26bd416eb4f5e9ee586118f995 Mon Sep 17 00:00:00 2001 From: DebaucheryLibrarian Date: Fri, 6 Nov 2026 01:10:06 +0100 Subject: [PATCH] Process users one at a time through the whole pipeline, added --queue [file] and --wait. Fixed self posts never being saved. Users now go through profile, posts, content info and downloads one by one (preparing the next user while the current one downloads), saving each index as soon as that user is done. The queue is re-read between users, users are removed once archived, and a lock file lets a second --queue invocation hand its users to the running process. Co-Authored-By: Claude Opus 5.5 --- .gitignore | 1 + CLAUDE.md | 19 ++- README.md | 2 + config/default.js | 9 ++ src/app.js | 268 +++++++++++++++++++++++++++++------ src/cli.js | 8 ++ src/fetch/content.js | 4 +- src/queue.js | 171 ++++++++++++++++++++++ src/sources/accountStatus.js | 32 ++--- src/sources/getUserPosts.js | 121 +++++++--------- src/utils/listFile.js | 79 +++++++++++ 11 files changed, 577 insertions(+), 137 deletions(-) create mode 100644 src/queue.js create mode 100644 src/utils/listFile.js diff --git a/.gitignore b/.gitignore index 019d0be..7ace3a8 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,4 @@ dist/ users* posts ignore +/queue* diff --git a/CLAUDE.md b/CLAUDE.md index fc09af0..2875cfe 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -17,13 +17,24 @@ Node.js CLI that archives reddit users' posts (and direct links to supported med ## Architecture -The pipeline runs from `src/app.js`: +`src/app.js` processes users **one at a time through the whole pipeline**, via `runUsers()`. While one user's media downloads, the next user is already being prepared (lookahead of one). Each user's index is saved as soon as that user is done. Do not go back to processing stage by stage across all users: that meant hours of reddit requests before the first download, everything lost on a crash, and no way to queue new users. -1. **Sources** (`src/sources/`): `getUserPosts` / `getPosts` fetch the user profile, social links and submissions through the reddit client, plus the existing index (`getIndex`, YAML at `library.index.file`). +Per user: + +1. **Sources** (`src/sources/`): `getUserPosts(username, tracker)` resolves to `{ status, user }`. Status is `active`, `empty`, `suspended`, `deleted` or `error`. It fetches the profile (`user.js`), the existing index (`getIndex`, YAML at `library.index.file`), social links if needed, and submissions. 2. **Curate** (`src/curate/`): `posts.js` normalizes reddit post objects, applies filters (index, dates, limit, ignore, include/exclude, duplicates) and runs `dissectLink` to choose a host method. `user.js` normalizes the profile. -3. **Info** (`src/fetch/info.js`): calls `methods[host.method]` (in `src/methods/`) to resolve each post into `{ album, items[] }`. If that fails, it falls back to reddit preview images (`redditPreview`). Some methods export `{ fetchInfo, fetchPredata }`; predata (e.g. a RedGifs temporary token) is fetched once per run. +3. **Info** (`src/fetch/info.js`): calls `methods[host.method]` (in `src/methods/`) to resolve each post into `{ album, items[] }`. If that fails, it falls back to reddit preview images (`redditPreview`). Some methods export `{ fetchInfo, fetchPredata }`. Predata (e.g. a RedGifs temporary token) is cached in `app.js` for `fetch.predataTtl`. 4. **Content** (`src/fetch/content.js`): downloads items (`src/fetch/item.js`, using bhttp), dedupes by blake2 hash, saves to a path built by `src/interpolate.js` from `config.library` patterns, muxes video+audio, writes EXIF and finally updates the index (`src/save/writeToIndex.js`). +Where users come from: `src/queue.js` provides `createListSource` (`--users`/`--file-users`) and `createQueueSource` (`--queue [file]`, default `queue`). `--posts` and `--fetch` still run as small batches. + +### Queue (`--queue [file]`, `--wait`) + +- The queue file is reread before each next user. A user is removed after finishing, including when suspended or deleted. Users with status `error` stay queued and are skipped for the rest of the run, so they can't loop. +- `.lock` (holding a pid, created with `wx`) ensures only one process runs a queue. A second `--queue --users …` process only appends its users and exits. Without `--queue`, the lock is ignored. +- At the end of a run, the lock is released first and the queue checked once more, which closes the race with a process that appended while it still saw the lock. +- List files are edited with `src/utils/listFile.js`: reread right before writing, only the matching lines touched, replaced atomically via rename. + ### Reddit client (`src/reddit/client.js`), the core of this branch - Calls reddit's public `.json` endpoints on `www.reddit.com` (`/user/x/about.json`, `/user/x/submitted.json`, `/comments/id.json`, `/api/morechildren.json`) with `raw_json=1`. There is no OAuth. @@ -42,7 +53,7 @@ The pipeline runs from `src/app.js`: ## Account status and cleanup - `src/sources/user.js` `fetchUser()` classifies accounts as `active`, `suspended` (about.json returns 200 with `is_suspended: true`; there is no date), `deleted` (JSON 404, which may also be a shadowban) or `error`. **`error` (network failure, 5xx, HTML bot block) must never count as dead**; only genuine JSON API errors carry `error.status` (see `requestJson`). -- Suspended and deleted users are skipped after that one request. `src/sources/accountStatus.js` records them in `users_dead.yml` (`config.users.dead`) with `firstSeen`. Once they have been dead for `retainDays[status]` (default 90), they are removed from the `--file-users` file, never from `--users`. A record is dropped if the account turns out active again. Files are only written at the end of a run. +- Suspended and deleted users are skipped after that one request. `src/sources/accountStatus.js` records them in `users_dead.yml` (`config.users.dead`) with `firstSeen`. Once they have been dead for `retainDays[status]` (default 90), they are removed from the `--file-users` file, never from `--users`. A record is dropped if the account turns out active again. The record is saved after every user. - `--cleanup` runs only this check (one request per user) without archiving. Regular archive runs do it too unless `users.dead.duringArchive` is false. - Social links (an extra HTML request) are only fetched when the profile description will actually be saved, i.e. it isn't indexed yet or `--redownload-profile` is set. diff --git a/README.md b/README.md index d40f867..994149d 100644 --- a/README.md +++ b/README.md @@ -46,6 +46,8 @@ ripunzel requires a arbitrarily recent version of Node.js. Before use, dependenc * `--posts [...]`: Fetch multiple posts by supplying a space-separated list of post IDs to `--posts`. * `--fetch `: Fetch content directly from an URL to an album or image on one of the supported hosts * `--file-users `: Fetch posts from multiple users by supplying a file with newline separated usernames +* `--queue []`: Work through a queue file (default `queue`, see `queue.file`) of newline separated usernames. Each user is removed from the file once archived (or found suspended/deleted), and the file is re-read before every next user, so users added while ripunzel is running are picked up along the way. Users given with `--users`/`--file-users` are added to the queue first. If another ripunzel process is already working through the same queue, they are only added for that process to pick up, and the new process exits. Users that fail (e.g. network errors) stay in the queue for the next run. +* `--wait`: With `--queue`, keep running once the queue is empty and check it for new users every `queue.interval` milliseconds (default 60 seconds). * `--cleanup`: Instead of archiving, only check the supplied users (`--users`/`--file-users`) for suspended or deleted accounts. These are recorded in `users_dead.yml` with the date they were first seen, and removed from the `--file-users` file once they've stayed dead for longer than `users.dead.retainDays` (90 days by default). Regular archive runs do the same unless `users.dead.duringArchive` is disabled. * `--file-posts `: Fetch multiple posts by supplying a file with newline separated post IDs * `--file-fetch `: Fetch content directly from multiple sources by supplying a file with newline separated URLs diff --git a/config/default.js b/config/default.js index d2179b4..ea117d5 100644 --- a/config/default.js +++ b/config/default.js @@ -70,6 +70,9 @@ module.exports = { watch: { schedule: '*/30 * * * *', }, + // how long (ms) per-method predata (e.g. RedGifs' temporary API token) is reused before + // being fetched again - matters for long --queue --wait runs + predataTtl: 60 * 60 * 1000, archives: { search: false, preview: true, @@ -115,6 +118,12 @@ module.exports = { minTime: 1500, }, }, + queue: { + // used by --queue when no filename is given + file: 'queue', + // with --queue --wait, how often (ms) an empty queue is checked for new users + interval: 60 * 1000, + }, users: { // record of suspended/deleted accounts, and pruning them from the --file-users list - see // src/sources/accountStatus.js. Also run on its own with --cleanup. diff --git a/src/app.js b/src/app.js index a192a01..2fae452 100644 --- a/src/app.js +++ b/src/app.js @@ -22,6 +22,10 @@ const { fetchSaveUserContent, fetchSaveDirectContent } = require('./fetch/conten const createAccountTracker = require('./sources/accountStatus'); const { fetchUser } = require('./sources/user'); +const { + acquireLock, releaseLock, createQueueSource, createListSource, +} = require('./queue'); +const { appendToList } = require('./utils/listFile'); const getPosts = require('./sources/getPosts')(reddit, args); const getUserPosts = require('./sources/getUserPosts')(reddit, args); @@ -30,7 +34,7 @@ async function getFileContents(location, label) { try { const fileContents = await fs.readFile(location, 'utf8'); - return fileContents.split('\n').filter((entry) => entry && entry.slice(0, 1) !== '#'); + return fileContents.split('\n').map((entry) => entry.trim()).filter((entry) => entry && entry.slice(0, 1) !== '#'); } catch (error) { logger.error(`Could not read ${label} file '${location}': ${error}.`); @@ -39,29 +43,35 @@ async function getFileContents(location, label) { } function getPostHosts(posts) { - // const hosts = Array.from(new Set(Object.values(posts).flatMap((user) => user.posts.map((post) => post.host?.method)))).filter(Boolean); const hosts = Object.values(Object.fromEntries(Object.values(posts).flatMap((user) => user.posts.map((post) => post.host && [post.host?.method, post.host])).filter(Boolean))); return hosts; } +// predata (e.g. RedGifs' temporary token, imgur's remaining credits) used to be fetched once per +// run, for all users at once. Users are now processed one at a time, and a --queue --wait run +// can go on indefinitely, so it's fetched when a method first needs it and cached for +// config.fetch.predataTtl - long enough not to refetch it for every user, short enough that +// a temporary token doesn't expire mid-run. +const predataCache = new Map(); + async function fetchPredata(hosts) { - return hosts.reduce(async (chain, host) => { - const acc = await chain; + const hostMethods = [...new Set(hosts.map((host) => host?.method))].filter((method) => methods[method]?.fetchPredata); - if (methods[host?.method]?.fetchPredata) { - const data = await methods[host.method].fetchPredata(); + return Promise.reduce(hostMethods, async (acc, method) => { + const cached = predataCache.get(method); - logger.info(`Fetched predata for ${host.method}`); - - return { - ...acc, - [host.method]: data, - }; + if (cached && Date.now() - cached.fetchedAt < config.fetch.predataTtl) { + return { ...acc, [method]: cached.data }; } - return acc; - }, Promise.resolve({})); + const data = await methods[method].fetchPredata(); + + predataCache.set(method, { data, fetchedAt: Date.now() }); + logger.info(`Fetched predata for ${method}`); + + return { ...acc, [method]: data }; + }, {}); } async function getUsernames() { @@ -100,52 +110,221 @@ async function cleanupUsers() { if ((index + 1) % 50 === 0) { logger.info(`Checked ${index + 1}/${usernames.length} accounts`); + await tracker.save(); } }); logger.info(`Checked ${usernames.length} accounts: ${statuses.active} active, ${statuses.suspended} suspended, ${statuses.deleted} deleted, ${statuses.error} failed to check`); await tracker.save(); + tracker.summarize(); } -async function getCompletePosts() { - let userPosts = {}; - let ignoreIds = []; - const { usernames, fileUsernames } = await getUsernames(); - let postIds = args.posts || []; +let ignoreIdsCache = null; - if (args.filePosts) { - postIds = postIds.concat(await getFileContents(args.filePosts, 'post ID')); +async function getIgnoreIds() { + if (!ignoreIdsCache) { + ignoreIdsCache = args.fileIgnore ? await getFileContents(args.fileIgnore, 'ignore') : []; } - if (!usernames.length && !postIds.length) { - return null; + return ignoreIdsCache; +} + +// everything up to (not including) downloading for one user: profile, posts, curation and +// content info. This is where nearly all the reddit requests happen. +async function prepareUser(username, tracker) { + try { + const { status, user } = await getUserPosts(username, tracker); + + if (!user) { + return { username, status, users: null }; + } + + const curated = curatePosts({ [user.name]: user }, await getIgnoreIds(), args); + const predata = await fetchPredata(getPostHosts(curated)); + const users = await attachContentInfo(curated, { reddit, predata }); + + return { username, status, users }; + } catch (error) { + logger.error(`Failed to fetch '${username}': ${args.debug ? error.stack : error.message}`); + + return { username, status: 'error', users: null }; + } +} + +async function saveUser(prepared, ep) { + if (!prepared.users) { + return prepared.status; } - if (usernames.length) { - const tracker = await createAccountTracker({ - usersFile: args.fileUsers, - fileUsernames, - enabled: config.users.dead.duringArchive, - }); + try { + await Promise.each(Object.values(prepared.users), (user) => fetchSaveUserContent(user, ep, args)); - userPosts = await getUserPosts(usernames, tracker); + return prepared.status; + } catch (error) { + logger.error(`Failed to save '${prepared.username}': ${args.debug ? error.stack : error.message}`); + return 'error'; + } +} + +function wait(ms) { + return new Promise((resolve) => { + setTimeout(resolve, ms); + }); +} + +// Processes users one at a time through the whole pipeline - profile, posts, content info, +// downloads, index - instead of the old stage-by-stage approach (every user's posts first, +// then every user's content info, then all downloads), which meant hours of reddit requests +// before the first download on a large users list, all of it lost on a crash, and no way to +// take on new users mid-run. Each user's index is now saved as soon as that user is done. +// +// While one user's media downloads (mostly bound by the hosting sites), the next user is +// already being prepared (mostly bound by reddit's throttle) - a lookahead of one, so the two +// kinds of waiting overlap without ever having more than two users in flight. +// +// source: see queue.js - next() hands out the next username (or null if there's none right +// now), done() is told how each one went. With `waitInterval`, an empty source is polled +// again after that many ms instead of ending the run (--queue --wait). +async function runUsers(source, ep, tracker, { waitInterval = null } = {}) { + const counts = {}; + let pending = null; + let waiting = false; + + const startNext = async () => { + const username = await source.next(); + + pending = username ? prepareUser(username, tracker) : null; + }; + + await startNext(); + + // eslint-disable-next-line no-constant-condition + while (true) { + if (!pending) { + if (!waitInterval) { + break; + } + + if (!waiting) { + logger.info(`Nothing left in ${source.label}, checking again every ${Math.round(waitInterval / 1000)}s`); + waiting = true; + } + + // eslint-disable-next-line no-await-in-loop + await wait(waitInterval); + // eslint-disable-next-line no-await-in-loop + await startNext(); + + // eslint-disable-next-line no-continue + continue; + } + + waiting = false; + + // eslint-disable-next-line no-await-in-loop + const prepared = await pending; + + // start preparing the next user before downloading this one's media + // eslint-disable-next-line no-await-in-loop + await startNext(); + + // eslint-disable-next-line no-await-in-loop + const status = await saveUser(prepared, ep); + + counts[status] = (counts[status] || 0) + 1; + + // eslint-disable-next-line no-await-in-loop + await source.done(prepared.username, status); + // eslint-disable-next-line no-await-in-loop await tracker.save(); } - if (postIds.length) { - userPosts = await getPosts(postIds, userPosts); + const total = Object.values(counts).reduce((acc, count) => acc + count, 0); + + if (total) { + logger.info(`Processed ${total} user(s) from ${source.label}: ${Object.entries(counts).map(([status, count]) => `${count} ${status}`).join(', ')}`); } - if (args.fileIgnore) { - ignoreIds = await getFileContents(args.fileIgnore, 'ignore'); + tracker.summarize(); +} + +// --queue [file]: queue the given users (if any), then work through the queue - unless another +// process already is, in which case it'll pick up the added users itself +async function runQueue(ep) { + const queueFile = args.queue || config.queue.file; + const { usernames, fileUsernames } = await getUsernames(); + + if (usernames.length) { + const added = await appendToList(queueFile, usernames); + + logger.info(`Added ${added.length} user(s) to queue '${queueFile}'${added.length < usernames.length ? ` (${usernames.length - added.length} already queued)` : ''}`); } - const curatedUserPosts = curatePosts(userPosts, ignoreIds, args); - const predata = await fetchPredata(getPostHosts(curatedUserPosts)); + const owner = await acquireLock(queueFile); - return attachContentInfo(curatedUserPosts, { reddit, predata }); + if (owner) { + logger.info(`Queue '${queueFile}' is already being processed by another ripunzel process (pid ${owner})${usernames.length ? ', which will pick up the added users' : ''}`); + + return; + } + + const tracker = await createAccountTracker({ + usersFile: args.fileUsers, + fileUsernames, + enabled: config.users.dead.duringArchive, + }); + + const source = createQueueSource(queueFile); + + logger.info(`Processing queue '${queueFile}'${args.wait ? ', waiting for new users when empty' : ''}`); + + await runUsers(source, ep, tracker, { waitInterval: args.wait ? config.queue.interval : null }); + + // another process may have appended users after this one's last look at the queue, but + // while it still held the lock - so that process left them to this one. Release first, + // then look again: anything appended before this check is seen here, anything after it + // finds no lock, so its own process takes over. + releaseLock(); + + if (await source.hasPending() && !(await acquireLock(queueFile))) { + await runUsers(source, ep, tracker); + releaseLock(); + } +} + +async function runUserList(ep) { + const { usernames, fileUsernames } = await getUsernames(); + + if (!usernames.length) { + return; + } + + const tracker = await createAccountTracker({ + usersFile: args.fileUsers, + fileUsernames, + enabled: config.users.dead.duringArchive, + }); + + await runUsers(createListSource(usernames), ep, tracker); +} + +// --posts/--file-posts: individual posts, grouped by author - small enough in practice to still +// run as one batch +async function runPosts(ep) { + const postIds = (args.posts || []).concat(args.filePosts ? await getFileContents(args.filePosts, 'post ID') : []); + + if (!postIds.length) { + return; + } + + const userPosts = await getPosts(postIds, {}); + const curated = curatePosts(userPosts, await getIgnoreIds(), args); + const predata = await fetchPredata(getPostHosts(curated)); + const users = await attachContentInfo(curated, { reddit, predata }); + + await Promise.each(Object.values(users), (user) => fetchSaveUserContent(user, ep, args)); } async function getDirectContent(links, ep) { @@ -185,11 +364,6 @@ async function getCompleteContents(ep) { return null; } -function fetchSavePosts(userPosts, ep) { - // don't map to apply concurrency limit and reduce network stress - return Promise.reduce(Object.values(userPosts), (acc, user) => fetchSaveUserContent(user, ep, args), null); -} - async function initApp() { let ep; @@ -207,12 +381,14 @@ async function initApp() { await getCompleteContents(ep); } - const userPosts = await getCompletePosts(); - - if (userPosts) { - await fetchSavePosts(userPosts, ep); + if (args.queue !== undefined) { + await runQueue(ep); + } else { + await runUserList(ep); } + await runPosts(ep); + if (args.watch) { logger.info(`Watch-mode enabled, checking again for new posts according to crontab '${config.fetch.watch.schedule}'.`); } diff --git a/src/cli.js b/src/cli.js index 2dc4323..3635e8c 100644 --- a/src/cli.js +++ b/src/cli.js @@ -31,6 +31,14 @@ function getArgs() { describe: 'Load reddit usernames from file', type: 'string', }) + .option('queue', { + describe: 'Work through a queue file of usernames (default: queue.file in config, \'queue\'), removing each user once archived and re-reading the file for new users as it goes. Users given with --users/--file-users are added to the queue. If another process is already working through the queue, only adds them and exits', + type: 'string', + }) + .option('wait', { + describe: 'With --queue: keep running when the queue is empty, checking it for new users every queue.interval ms', + type: 'boolean', + }) .option('cleanup', { describe: 'Only check the supplied users for suspended/deleted accounts: record them, and remove them from --file-users once dead for longer than configured. Does not archive anything', type: 'boolean', diff --git a/src/fetch/content.js b/src/fetch/content.js index 82a8541..80c06e6 100644 --- a/src/fetch/content.js +++ b/src/fetch/content.js @@ -51,8 +51,10 @@ function selfPostToText(item, post) { async function getBuffers(item, context) { if (item.self) { + // `buffer:`, not a spread - spreading a Buffer copies its bytes into numbered keys, + // leaving `.buffer` undefined, so self posts were silently never saved return [{ - ...Buffer.from(selfPostToText(item, context.post), 'utf8'), + buffer: Buffer.from(selfPostToText(item, context.post), 'utf8'), hash: context.post.hash, }]; } diff --git a/src/queue.js b/src/queue.js new file mode 100644 index 0000000..2237da8 --- /dev/null +++ b/src/queue.js @@ -0,0 +1,171 @@ +'use strict'; + +const fs = require('fs-extra'); + +const { readList, removeFromList } = require('./utils/listFile'); +const logger = require('./logger')(__filename); + +// --queue [file]: a users list that's consumed as it's processed - each user is removed once +// archived (or found suspended/deleted), and the file is re-read before every next user, so +// users added while ripunzel is running are picked up without restarting it. A lock file +// (.lock, holding the pid) makes sure only one process works through a queue: another +// `--queue --users ...` invocation just appends its users and leaves them to the running one. + +function lockPath(queueFile) { + return `${queueFile}.lock`; +} + +function isAlive(pid) { + try { + process.kill(pid, 0); + + return true; + } catch (error) { + // EPERM: exists, but owned by someone else + return error.code === 'EPERM'; + } +} + +async function lockOwner(queueFile) { + try { + const pid = Number(await fs.readFile(lockPath(queueFile), 'utf8')); + + return pid && isAlive(pid) ? pid : null; + } catch (error) { + return null; + } +} + +let heldLock = null; + +function releaseLock() { + if (!heldLock) { + return; + } + + try { + // only remove it if it's still ours + if (Number(fs.readFileSync(heldLock, 'utf8')) === process.pid) { + fs.unlinkSync(heldLock); + } + } catch (error) { + // already gone + } + + heldLock = null; +} + +process.on('exit', releaseLock); + +['SIGINT', 'SIGTERM'].forEach((signal) => { + process.once(signal, () => { + releaseLock(); + process.exit(signal === 'SIGINT' ? 130 : 143); + }); +}); + +// returns null when acquired, or the pid of the live process holding it +async function acquireLock(queueFile) { + const file = lockPath(queueFile); + + try { + // 'wx' fails if it exists - atomic, so two processes starting at once can't both get it + await fs.writeFile(file, String(process.pid), { flag: 'wx' }); + } catch (error) { + if (error.code !== 'EEXIST') { + throw error; + } + + const owner = await lockOwner(queueFile); + + if (owner) { + return owner; + } + + logger.warn(`Removing stale queue lock '${file}' (process no longer running)`); + + await fs.remove(file); + + return acquireLock(queueFile); + } + + heldLock = file; + + return null; +} + +// a source of usernames for app.js's runUsers(): next() hands out the next user not yet taken +// in this run (re-reading the file every time), done() removes it from the queue unless it +// failed - a failed user stays queued for the next run, but isn't retried in this one, so a +// persistent failure can't loop forever +function createQueueSource(queueFile) { + const taken = new Set(); + + return { + label: `queue '${queueFile}'`, + async next() { + const entries = await readList(queueFile); + const username = entries.find((entry) => !taken.has(entry.toLowerCase())); + + if (username) { + taken.add(username.toLowerCase()); + } + + return username || null; + }, + async done(username, status) { + if (status === 'error') { + logger.warn(`Leaving '${username}' in ${this.label} to retry on the next run`); + + return; + } + + await removeFromList(queueFile, [username]); + + const remaining = (await readList(queueFile)).filter((entry) => !taken.has(entry.toLowerCase())); + + logger.info(`Finished '${username}' (${status}), ${remaining.length} user(s) left in ${this.label}`); + }, + // see app.js's runQueue(): closes the gap between deciding the queue is empty and + // another process appending to it while it still saw this one's lock + async hasPending() { + const entries = await readList(queueFile); + + return entries.some((entry) => !taken.has(entry.toLowerCase())); + }, + }; +} + +// plain --users/--file-users: a fixed list, nothing to remove afterwards (the dead account +// tracker takes care of the users file) +function createListSource(usernames) { + const seen = new Set(); + const pending = usernames.filter((username) => { + const lower = username.toLowerCase(); + + if (seen.has(lower)) { + return false; + } + + seen.add(lower); + + return true; + }); + + return { + label: 'users', + async next() { + return pending.shift() || null; + }, + async done(username, status) { + logger.verbose(`Finished '${username}' (${status}), ${pending.length} user(s) left`); + }, + }; +} + +module.exports = { + acquireLock, + releaseLock, + createQueueSource, + createListSource, +}; diff --git a/src/sources/accountStatus.js b/src/sources/accountStatus.js index fa73bf8..93c43c1 100644 --- a/src/sources/accountStatus.js +++ b/src/sources/accountStatus.js @@ -4,6 +4,8 @@ const config = require('config'); const fs = require('fs-extra'); const yaml = require('js-yaml'); +const { removeFromList } = require('../utils/listFile'); + const logger = require('../logger')(__filename); const DAY = 24 * 60 * 60 * 1000; @@ -22,7 +24,8 @@ const DAY = 24 * 60 * 60 * 1000; // lifted suspension or shadowban is noticed in time. After that it's removed from the users // file - but only if it came from one (--file-users), never from --users on the command line. // An account that turns out active again has its record dropped. Nothing is written until -// save(), at the end of the run, since users are processed concurrently. +// save(), which is called after every user - a long (e.g. --queue) run shouldn't lose its +// observations when interrupted. async function loadRecords(file) { try { return yaml.load(await fs.readFile(file, 'utf8')) || {}; @@ -41,17 +44,6 @@ function findKey(records, username) { return Object.keys(records).find((key) => key.toLowerCase() === lower); } -async function removeFromUsersFile(usersFile, usernames) { - const lower = new Set([...usernames].map((name) => name.toLowerCase())); - const contents = await fs.readFile(usersFile, 'utf8'); - // keeps comments, blank lines and order intact - only drops the exact matching entries - const kept = contents.split('\n').filter((line) => !lower.has(line.trim().toLowerCase())); - const tmpFile = `${usersFile}.tmp`; - - await fs.writeFile(tmpFile, kept.join('\n')); - await fs.rename(tmpFile, usersFile); -} - async function createAccountTracker({ usersFile = null, fileUsernames = [], enabled = true } = {}) { const deadConfig = config.users?.dead || {}; const file = deadConfig.file || 'users_dead.yml'; @@ -122,16 +114,22 @@ async function createAccountTracker({ usersFile = null, fileUsernames = [], enab } if (removals.size && usersFile) { - await removeFromUsersFile(usersFile, removals); - counts.removed = removals.size; + await removeFromList(usersFile, removals); + counts.removed += removals.size; + removals.clear(); } await fs.writeFile(file, yaml.dump(records, { sortKeys: true })); - - logger.info(`Dead account record '${file}' updated: ${counts.suspended} suspended, ${counts.deleted} deleted, ${counts.restored} active again, ${counts.removed} removed from users file`); + changed = false; } - return { record, save }; + function summarize() { + if (enabled && (counts.suspended || counts.deleted || counts.restored)) { + logger.info(`Dead account record '${file}': ${counts.suspended} suspended, ${counts.deleted} deleted, ${counts.restored} active again, ${counts.removed} removed from users file`); + } + } + + return { record, save, summarize }; } module.exports = createAccountTracker; diff --git a/src/sources/getUserPosts.js b/src/sources/getUserPosts.js index 8fb87a1..1876465 100644 --- a/src/sources/getUserPosts.js +++ b/src/sources/getUserPosts.js @@ -1,7 +1,5 @@ 'use strict'; -const Promise = require('bluebird'); - const getIndex = require('./getIndex'); const getArchivePostIds = require('../archives/getArchivePostIds'); const { fetchUser, indexUser, resolveUser } = require('./user'); @@ -45,7 +43,8 @@ async function getPosts(username, reddit, args, until) { } catch (error) { logger.error(`Failed to fetch posts from reddit user '${username}': ${error.message} (https://reddit.com/user/${username})`); - return []; + // distinct from [] (no posts), so the caller can tell a failure apart - see below + return null; } } @@ -55,76 +54,60 @@ async function getArchivedPosts(username, posts, reddit) { return Promise.all(postIds.map((postId) => reddit.getSubmission(postId))); } +// fetches a single user's profile and posts. Resolves to { status, user }: +// - status 'active' with user (incl. posts and index) when there's something to curate, +// 'empty' when the account has no posts, 'suspended'/'deleted' (see sources/user.js), or +// 'error' when neither the profile nor the posts could be fetched - the latter is what +// e.g. the queue uses to decide a user should be retried later instead of being dropped +// - tracker: see accountStatus.js - records suspended/deleted accounts, optional function getUserPostsWrap(reddit, args) { - // tracker: see accountStatus.js - records suspended/deleted accounts (and drops them from the - // users file once they've been dead long enough), optional - return async function getUserPosts(usernames, tracker = null) { - try { - const users = await Promise.map(usernames, async (username) => { - try { - // sequential rather than parallel: the user decides whether there's anything to - // fetch at all, the index (whose path may depend on the user's details) - // decides how far back to page and whether social links are needed, and - // every reddit request goes through the same single-concurrency limiter anyway - const fetched = await fetchUser(username, reddit); + return async function getUserPosts(username, tracker = null) { + // sequential: the user decides whether there's anything to fetch at all, the index + // (whose path may depend on the user's details) decides how far back to page and + // whether social links are needed, and every reddit request goes through the same + // single-concurrency limiter anyway + const fetched = await fetchUser(username, reddit); - tracker?.record(username, fetched.status); + tracker?.record(username, fetched.status); - // nothing more to get from a suspended or deleted account - its listing is a - // 403/404 too, and skipping it saves the rest of this user's requests - if (fetched.status === 'suspended' || fetched.status === 'deleted') { - return null; - } - - if (fetched.status === 'active') { - logger.info(`Fetched user profile for '${username}' (https://reddit.com/user/${username})`); - } - - const { profile, posts: indexed } = await getIndex(indexUser(username, fetched)); - const user = await resolveUser(username, fetched, profile, reddit, args); - const posts = await getPosts(username, reddit, args, getCutoff(indexed, args)); - - if (args.archives) { - posts.push(...await getArchivedPosts(username, posts, reddit)); - } - - if (posts.length) { - return { - ...user, - posts, - indexed: { - profile, - original: indexed, - updated: [], - }, - }; - } - - return null; - } catch (error) { - logger.error(`Failed to fetch posts from '${username}': ${error.message}`); - - return null; - } - }, { - concurrency: 10, - }); - - return users.reduce( - (userPosts, user) => (user - ? { - ...userPosts, - [user.name]: user, - } - : userPosts - ), - {}, - ); - } catch (error) { - logger.error(error); - - throw error; + // nothing more to get from a suspended or deleted account - its listing is a 403/404 + // too, and skipping it saves the rest of this user's requests + if (fetched.status === 'suspended' || fetched.status === 'deleted') { + return { status: fetched.status, user: null }; } + + if (fetched.status === 'active') { + logger.info(`Fetched user profile for '${username}' (https://reddit.com/user/${username})`); + } + + const { profile, posts: indexed } = await getIndex(indexUser(username, fetched)); + const user = await resolveUser(username, fetched, profile, reddit, args); + const posts = await getPosts(username, reddit, args, getCutoff(indexed, args)); + + if (posts === null) { + return { status: 'error', user: null }; + } + + if (args.archives) { + posts.push(...await getArchivedPosts(username, posts, reddit)); + } + + if (!posts.length) { + return { status: 'empty', user: null }; + } + + return { + status: 'active', + user: { + ...user, + posts, + indexed: { + profile, + original: indexed, + updated: [], + }, + }, + }; }; } diff --git a/src/utils/listFile.js b/src/utils/listFile.js new file mode 100644 index 0000000..35f0556 --- /dev/null +++ b/src/utils/listFile.js @@ -0,0 +1,79 @@ +'use strict'; + +const fs = require('fs-extra'); + +// helpers for the plain newline-separated list files (users, queue): one entry per line, '#' +// comments and blank lines allowed. Writes only ever touch the matching entries, keep everything +// else (order, comments) as-is, and replace the file atomically via rename - the file may be +// edited by hand while ripunzel is running, so it's always re-read right before writing to keep +// that window as small as possible. + +function parseList(contents) { + return contents + .split('\n') + .map((line) => line.trim()) + .filter((line) => line && !line.startsWith('#')); +} + +async function readList(file) { + try { + return parseList(await fs.readFile(file, 'utf8')); + } catch (error) { + if (error.code === 'ENOENT') { + return []; + } + + throw error; + } +} + +async function writeAtomic(file, contents) { + const tmpFile = `${file}.${process.pid}.tmp`; + + await fs.writeFile(tmpFile, contents); + await fs.rename(tmpFile, file); +} + +// case-insensitive, like reddit usernames +async function removeFromList(file, entries) { + const remove = new Set([...entries].map((entry) => entry.toLowerCase())); + const contents = await fs.readFile(file, 'utf8'); + const kept = contents.split('\n').filter((line) => !remove.has(line.trim().toLowerCase())); + + await writeAtomic(file, kept.join('\n')); +} + +// appends the entries not already listed, returns the ones actually added +async function appendToList(file, entries) { + const contents = await fs.readFile(file, 'utf8').catch((error) => { + if (error.code === 'ENOENT') { + return ''; + } + + throw error; + }); + + const present = new Set(parseList(contents).map((entry) => entry.toLowerCase())); + const added = []; + + entries.forEach((entry) => { + if (!present.has(entry.toLowerCase())) { + present.add(entry.toLowerCase()); + added.push(entry); + } + }); + + if (added.length) { + const separator = contents && !contents.endsWith('\n') ? '\n' : ''; + + await writeAtomic(file, `${contents}${separator}${added.join('\n')}\n`); + } + + return added; +} + +module.exports = { + readList, + removeFromList, + appendToList, +};