Benature/omnivore
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge all changes from main, update theming of Discover

Browse Source
This commit is contained in:
Thomas Rogers
2024-03-07 17:39:57 +01:00
parent 451939832f ee5000393f
commit 7d291f084b
691 changed files with 82404 additions and 41037 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
packages/import-handler/.eslintrc
View File

@ -4,6 +4,7 @@
"project": "tsconfig.json"
},
"rules": {
"@typescript-eslint/no-explicit-any": "warn"
"@typescript-eslint/no-explicit-any": "warn",
" @typescript-eslint/no-unnecessary-type-assertion": "off"
}
}

5
packages/import-handler/package.json
View File

@ -13,10 +13,9 @@
"test:typecheck": "tsc --noEmit",
"lint": "eslint src --ext ts,js,tsx,jsx",
"compile": "tsc",
"build": "tsc && yarn copy-files",
"build": "tsc",
"start": "functions-framework --target=importHandler",
"dev": "concurrently \"tsc -w\" \"nodemon --watch ./build/ --exec npm run start\"",
"copy-files": "copyfiles src/luaScripts/*.lua build/",
"start:collector": "functions-framework --target=importMetricsCollector"
},
"devDependencies": {
@ -45,10 +44,10 @@
"dompurify": "^2.4.3",
"fs-extra": "^11.1.0",
"glob": "^8.1.0",
"ioredis": "^5.3.2",
"jsonwebtoken": "^8.5.1",
"linkedom": "^0.14.21",
"nodemon": "^2.0.15",
"redis": "^4.3.1",
"unzip-stream": "^0.3.1",
"urlsafe-base64": "^1.0.0",
"uuid": "^9.0.0"

13
packages/import-handler/src/index.ts
View File

@ -2,10 +2,10 @@ import { Storage } from '@google-cloud/storage'
import { Readability } from '@omnivore/readability'
import * as Sentry from '@sentry/serverless'
import axios from 'axios'
import Redis from 'ioredis'
import * as jwt from 'jsonwebtoken'
import { Stream } from 'node:stream'
import * as path from 'path'
import { createClient } from 'redis'
import { promisify } from 'util'
import { v4 as uuid } from 'uuid'
import { importCsv } from './csv'
@ -14,9 +14,6 @@ import { ImportStatus, updateMetrics } from './metrics'
import { createRedisClient } from './redis'
import { CONTENT_FETCH_URL, createCloudTask, emailUserUrl } from './task'
// explicitly create the return type of RedisClient
type RedisClient = ReturnType<typeof createClient>
export enum ArticleSavingRequestStatus {
Failed = 'FAILED',
Processing = 'PROCESSING',
@ -59,7 +56,7 @@ export type ImportContext = {
countFailed: number
urlHandler: UrlHandler
contentHandler: ContentHandler
redisClient: RedisClient
redisClient: Redis
taskId: string
source: string
}
@ -300,7 +297,7 @@ const contentHandler = async (
return Promise.resolve()
}
const handleEvent = async (data: StorageEvent, redisClient: RedisClient) => {
const handleEvent = async (data: StorageEvent, redisClient: Redis) => {
if (shouldHandle(data)) {
const handler = handlerForFile(data.name)
if (!handler) {
@ -367,7 +364,7 @@ export const importHandler = Sentry.GCPFunction.wrapHttpFunction(
const obj = getStorageEvent(pubSubMessage)
if (obj) {
// create redis client
const redisClient = await createRedisClient(
const redisClient = createRedisClient(
process.env.REDIS_URL,
process.env.REDIS_CERT
)
@ -416,7 +413,7 @@ export const importMetricsCollector = Sentry.GCPFunction.wrapHttpFunction(
return res.status(400).send('Bad Request')
}
const redisClient = await createRedisClient(
const redisClient = createRedisClient(
process.env.REDIS_URL,
process.env.REDIS_CERT
)

34
packages/import-handler/src/luaScripts/updateMetrics.lua
View File

@ -1,34 +0,0 @@
local key = tostring(KEYS[1]);
local status = tostring(ARGV[1]);
local timestamp = tonumber(ARGV[2]);
-- increment the status counter
redis.call('HINCRBY', key, status, 1);
if (status == "imported" or status == "failed") then
-- get the current metrics
local bulk = redis.call('HGETALL', key);
-- get the total, imported and failed counters
local result = {}
local nextkey
for i, v in ipairs(bulk) do
if i % 2 == 1 then
nextkey = v
else
result[nextkey] = v
end
end
local imported = tonumber(result['imported']) or 0;
local failed = tonumber(result['failed']) or 0;
local total = tonumber(result['total']) or 0;
local state = tonumber(result['state']) or 0;
if (state == 0 and imported + failed >= total) then
-- all the records have been processed
-- update the metrics
redis.call('HSET', key, 'end_time', timestamp, 'state', 1);
return 1
end
end
return 0;

70
packages/import-handler/src/metrics.ts
View File

@ -1,9 +1,5 @@
import { createClient } from 'redis'
import Redis from 'ioredis'
import { sendImportCompletedEmail } from '.'
import { lua } from './redis'
// explicitly create the return type of RedisClient
type RedisClient = ReturnType<typeof createClient>
export enum ImportStatus {
STARTED = 'started',
@ -31,7 +27,7 @@ interface ImportMetrics {
}
export const createMetrics = async (
redisClient: RedisClient,
redisClient: Redis,
userId: string,
taskId: string,
source: string
@ -39,7 +35,7 @@ export const createMetrics = async (
const key = `import:${userId}:${taskId}`
try {
// set multiple fields
await redisClient.hSet(key, {
await redisClient.hset(key, {
['start_time']: Date.now(),
['source']: source,
['state']: ImportTaskState.STARTED,
@ -50,22 +46,66 @@ export const createMetrics = async (
}
export const updateMetrics = async (
redisClient: RedisClient,
redisClient: Redis,
userId: string,
taskId: string,
status: ImportStatus
) => {
const key = `import:${userId}:${taskId}`
/**
* Define our command
*/
redisClient.defineCommand('updatemetrics', {
numberOfKeys: 1,
lua: `
local key = tostring(KEYS[1]);
local status = tostring(ARGV[1]);
local timestamp = tonumber(ARGV[2]);
-- increment the status counter
redis.call('HINCRBY', key, status, 1);
if (status == "imported" or status == "failed") then
-- get the current metrics
local bulk = redis.call('HGETALL', key);
-- get the total, imported and failed counters
local result = {}
local nextkey
for i, v in ipairs(bulk) do
if i % 2 == 1 then
nextkey = v
else
result[nextkey] = v
end
end
local imported = tonumber(result['imported']) or 0;
local failed = tonumber(result['failed']) or 0;
local total = tonumber(result['total']) or 0;
local state = tonumber(result['state']) or 0;
if (state == 0 and imported + failed >= total) then
-- all the records have been processed
-- update the metrics
redis.call('HSET', key, 'end_time', timestamp, 'state', 1);
return 1
end
end
return 0;
`,
})
try {
// use lua script to increment hash field
const state = await redisClient.evalSha(lua.sha, {
keys: [key],
arguments: [status, Date.now().toString()],
})
const state = await redisClient.updatemetrics(
key,
status,
Date.now().toString()
)
// if the task is finished, send email
if (state == ImportTaskState.FINISHED) {
if ((state as ImportTaskState) == ImportTaskState.FINISHED) {
const metrics = await getMetrics(redisClient, userId, taskId)
if (metrics) {
await sendImportCompletedEmail(userId, metrics.imported, metrics.failed)
@ -77,13 +117,13 @@ export const updateMetrics = async (
}
export const getMetrics = async (
redisClient: RedisClient,
redisClient: Redis,
userId: string,
taskId: string
): Promise<ImportMetrics | null> => {
const key = `import:${userId}:${taskId}`
try {
const metrics = await redisClient.hGetAll(key)
const metrics = await redisClient.hgetall(key)
return {
// convert to integer

76
packages/import-handler/src/redis.ts
View File

@ -1,41 +1,45 @@
import { readFileSync } from 'fs'
import path from 'path'
import { createClient } from 'redis'
import { Redis, Result } from 'ioredis'
// load lua script
export const lua = {
script: readFileSync(
path.resolve(__dirname, 'luaScripts/updateMetrics.lua'),
'utf8'
),
sha: '',
}
export const createRedisClient = async (url?: string, cert?: string) => {
const redisClient = createClient({
url,
socket: {
tls: url?.startsWith('rediss://'), // rediss:// is the protocol for TLS
cert: cert?.replace(/\\n/g, '\n'), // replace \n with new line
rejectUnauthorized: false, // for self-signed certs
connectTimeout: 10000, // 10 seconds
reconnectStrategy(retries: number): number | Error {
if (retries > 10) {
return new Error('Retries exhausted')
export const createRedisClient = (url?: string, cert?: string) => {
return new Redis(url || 'redis://localhost:6379', {
connectTimeout: 10000, // 10 seconds
tls: cert
? {
cert: cert.replace(/\\n/g, '\n'), // replace \n with new line
rejectUnauthorized: false, // for self-signed certs
}
return 1000
},
: undefined,
reconnectOnError: (err) => {
const targetErrors = [/READONLY/, /ETIMEDOUT/]
targetErrors.forEach((targetError) => {
if (targetError.test(err.message)) {
// Only reconnect when the error contains the keyword
return true
}
})
return false
},
retryStrategy: (times) => {
if (times > 10) {
// End reconnecting after a specific number of tries and flush all commands with a individual error
return null
}
// reconnect after
return Math.min(times * 50, 2000)
},
})
redisClient.on('error', (err) => console.error('Redis Client Error', err))
await redisClient.connect()
console.log('Redis Client Connected:', url)
// load script to redis
lua.sha = await redisClient.scriptLoad(lua.script)
console.log('Redis Lua Script Loaded', lua.sha)
return redisClient
}
// Add declarations
declare module 'ioredis' {
interface RedisCommander<Context> {
updatemetrics(
key: string,
arg1: string,
arg2: string
): Result<number, Context>
}
}