|
|
|
@@ -0,0 +1,393 @@
|
|
|
|
|
/** Gitea source kind: notification, stream and thread polling, thread context, and the Cordis plugin entry (see GITEA-DESIGN.md). */
|
|
|
|
|
import type { Context } from '@deepseek-ai/cordis'
|
|
|
|
|
import { z } from 'zod'
|
|
|
|
|
import { sameName } from '../infra/text.ts'
|
|
|
|
|
import { GiteaSetupSchema } from '../deployment.ts'
|
|
|
|
|
import { validItems, type Tally } from '../infra/commands.ts'
|
|
|
|
|
import { KV } from '../kv.ts'
|
|
|
|
|
import { giteaClient, GiteaApiError, type GiteaApi } from './gitea-api.ts'
|
|
|
|
|
import { GiteaCommentSchema, GiteaIssueSchema, GiteaReviewSchema, SOURCE, commentNumber, commentTrigger, giteaAddress, issueTrigger, reviewTrigger,
|
|
|
|
|
type GiteaComment, type GiteaIssue, type GiteaRef } from './gitea-trigger.ts'
|
|
|
|
|
import { giteaLink, sendGitea } from './gitea-channel.ts'
|
|
|
|
|
import { health, sourceContext, type Health, type SourceContext } from './host.ts'
|
|
|
|
|
import { cursorAt, pause, POLL_HORIZON_MS, pollThreads as pollEach } from './poll.ts'
|
|
|
|
|
import { pollFailure } from './github.ts'
|
|
|
|
|
import type { ContextMessage, SourceKind, ThreadContext } from './kind.ts'
|
|
|
|
|
import type { Thread } from '../core/index.ts'
|
|
|
|
|
import type { Retry } from './github.ts'
|
|
|
|
|
|
|
|
|
|
export const GiteaConfigSchema = GiteaSetupSchema.extend({
|
|
|
|
|
pollSeconds: z.number().int().min(60).max(3600).default(60),
|
|
|
|
|
failuresBeforeError: z.number().int().min(1).max(20).default(3),
|
|
|
|
|
observeIssues: z.boolean().default(false), observePulls: z.boolean().default(false), observeComments: z.boolean().default(false),
|
|
|
|
|
}).strict()
|
|
|
|
|
export type GiteaConfig = z.infer<typeof GiteaConfigSchema>
|
|
|
|
|
const OBSERVED = { issues: 'observeIssues', pulls: 'observePulls', comments: 'observeComments' } as const
|
|
|
|
|
type Scope = keyof typeof OBSERVED
|
|
|
|
|
function scopeOf(meta: Record<string, unknown>): Scope | undefined {
|
|
|
|
|
if (meta.kind === 'issue') return 'issues'
|
|
|
|
|
if (meta.kind === 'pull') return 'pulls'
|
|
|
|
|
return meta.kind === 'conversation' || meta.kind === 'review' ? 'comments' : undefined
|
|
|
|
|
}
|
|
|
|
|
/** The notification of a Gitea thread; `subject.url` is the API address of its Issue or PR. */
|
|
|
|
|
const NotificationSchema = z.looseObject({
|
|
|
|
|
repository: z.looseObject({ full_name: z.string() }), subject: z.looseObject({ url: z.string().nullish() }),
|
|
|
|
|
})
|
|
|
|
|
/** Reads the Issue body and the conversation of an Issue or PR for assessments; Gitea comment threads are flat. */
|
|
|
|
|
export const giteaThreadContext: ThreadContext<GiteaConfig> = {
|
|
|
|
|
messageId: trigger => trigger.meta.kind === 'issue' || trigger.meta.kind === 'pull' ? `${trigger.meta.number}-body`
|
|
|
|
|
: trigger.meta.kind === 'review' ? `r${trigger.meta.reviewId}` : `c${trigger.meta.commentId}`,
|
|
|
|
|
read: async (_commands, trigger, part, cursor, config) => {
|
|
|
|
|
const number = Number(trigger.meta.number)
|
|
|
|
|
if (!number) throw new Error('Gitea thread has no Issue or PR number')
|
|
|
|
|
if (!config.url) throw new Error('The Gitea Source names no URL')
|
|
|
|
|
const api = giteaClient(config.url, config.tokenEnv)
|
|
|
|
|
const issue = await api.json(GiteaIssueSchema, `repos/${config.repository}/issues/${number}`)
|
|
|
|
|
const bodyMessage: ContextMessage = { id: `${number}-body`, from: issue.user.login, at: issue.created_at,
|
|
|
|
|
text: `${issue.title}\n\n${issue.body ?? ''}`, ...(issue.html_url ? { url: issue.html_url } : {}) }
|
|
|
|
|
if (part === 'root' || part === 'parent') return { messages: [bodyMessage], body: { messages: [bodyMessage] } }
|
|
|
|
|
const comments = await api.pages(z.array(GiteaCommentSchema), `repos/${config.repository}/issues/${number}/comments`)
|
|
|
|
|
const messages: ContextMessage[] = [bodyMessage, ...comments.map(comment => ({
|
|
|
|
|
id: `c${comment.id}`, from: comment.user.login, at: comment.updated_at ?? comment.created_at ?? issue.updated_at ?? '',
|
|
|
|
|
text: comment.body ?? '', ...(comment.html_url ? { url: comment.html_url } : {}),
|
|
|
|
|
})).sort((a, b) => Date.parse(a.at) - Date.parse(b.at))]
|
|
|
|
|
if (issue.pull_request) {
|
|
|
|
|
const reviews = await api.pages(z.array(GiteaReviewSchema), `repos/${config.repository}/pulls/${number}/reviews`)
|
|
|
|
|
for (const review of reviews) if (review.body) messages.push({ id: `r${review.id}`, from: review.user.login,
|
|
|
|
|
at: review.submitted_at ?? '', text: review.body, ...(review.html_url ? { url: review.html_url } : {}) })
|
|
|
|
|
messages.sort((a, b) => Date.parse(a.at) - Date.parse(b.at))
|
|
|
|
|
}
|
|
|
|
|
const page = Math.max(1, Number(/^p(\d+)$/.exec(cursor ?? '')?.[1] ?? 1) || 1)
|
|
|
|
|
const items = messages.slice((page - 1) * 100, page * 100)
|
|
|
|
|
return { messages: items, ...(messages.length > page * 100 ? { next: `p${page + 1}` } : {}), body: { messages: items } }
|
|
|
|
|
},
|
|
|
|
|
}
|
|
|
|
|
export const giteaKind: SourceKind<GiteaConfig> = {
|
|
|
|
|
kind: SOURCE,
|
|
|
|
|
title: { zh: 'Gitea', en: 'Gitea' },
|
|
|
|
|
description: { zh: '智能体的 Gitea 账号；接收这个 Gitea 实例发给它的通知和已订阅 Issue 与 PR 的讨论。',
|
|
|
|
|
en: 'The agent\'s Gitea account; receives the notifications this Gitea instance sends it and the discussion on subscribed Issues and PRs.' },
|
|
|
|
|
schema: GiteaConfigSchema,
|
|
|
|
|
fields: [
|
|
|
|
|
{ key: 'url', type: 'text', pattern: 'https?://[^\\/]+', label: { zh: 'Gitea 实例地址', en: 'Gitea instance' },
|
|
|
|
|
hint: { zh: '不带末尾斜杠，例如 https://gitea.example.com。保存时检查智能体账号能否推送到仓库，改动后重启 dsh-superagent。',
|
|
|
|
|
en: 'Without a trailing slash, for example https://gitea.example.com. Saving checks that the agent account can push to the repository; changing it restarts dsh-superagent.' } },
|
|
|
|
|
{ key: 'repository', type: 'text', pattern: '[A-Za-z0-9_.\-]+/[A-Za-z0-9_.\-]+', label: { zh: '监听的仓库', en: 'Repository' },
|
|
|
|
|
hint: { zh: '写作 owner/name。这个来源接收这个仓库的通知和讨论。', en: 'Written owner/name. This Source receives the repository\'s notifications and discussion.' } },
|
|
|
|
|
{ key: 'tokenEnv', type: 'text', pattern: '[A-Za-z_][A-Za-z0-9_]*', label: { zh: '智能体令牌的变量名', en: 'Token variable' },
|
|
|
|
|
hint: { zh: '存放智能体 Gitea API 令牌的环境变量名称，令牌本身不写进配置；建议放进取凭据文件（DSH 会注入进程环境）。',
|
|
|
|
|
en: 'Name of the environment variable holding the agent\'s Gitea API token; the token itself is never stored in configuration. Prefer the credentials file, which DSH injects into the process environment.' } },
|
|
|
|
|
{ key: 'pollSeconds', type: 'number', advanced: true, min: 60, max: 3600, label: { zh: '轮询间隔 / 秒', en: 'Polling interval / seconds' },
|
|
|
|
|
hint: { zh: '实例要求更长间隔时以实例为准。', en: 'A longer interval requested by the instance wins.' } },
|
|
|
|
|
{ key: 'failuresBeforeError', type: 'number', advanced: true, min: 1, max: 20, label: { zh: '连续失败几次后报告异常', en: 'Failed polls before reporting an error' },
|
|
|
|
|
hint: { zh: '单次网络中断会在下一次轮询时自动重试，不必报告。', en: 'A single network interruption is retried on the next poll and need not be reported.' } },
|
|
|
|
|
{ key: 'observeIssues', type: 'switch', label: { zh: '接收仓库里的新 Issue', en: 'Receive new Issues in the repository' },
|
|
|
|
|
hint: { zh: '打开后，没有 @ 智能体的新 Issue 也会收到，由策略的接收表决定是否成为候选。从打开的时候开始，不读以前的。',
|
|
|
|
|
en: 'New Issues that do not mention the agent are received too, and the Policy\'s admission table decides whether they become candidates. Starts when switched on; earlier ones are not read.' } },
|
|
|
|
|
{ key: 'observePulls', type: 'switch', label: { zh: '接收仓库里的新 PR', en: 'Receive new pull requests in the repository' },
|
|
|
|
|
hint: { zh: '打开后，没有 @ 智能体的新 PR 也会收到，由策略的接收表决定是否成为候选。',
|
|
|
|
|
en: 'New pull requests that do not mention the agent are received too, and the Policy\'s admission table decides whether they become candidates.' } },
|
|
|
|
|
{ key: 'observeComments', type: 'switch', label: { zh: '接收仓库里的评论', en: 'Receive comments in the repository' },
|
|
|
|
|
hint: { zh: '打开后，Issue 和 PR 上没有 @ 智能体的评论也会收到，由策略的接收表决定是否成为候选。',
|
|
|
|
|
en: 'Comments on Issues and PRs that do not mention the agent are received too, and the Policy\'s admission table decides whether they become candidates.' } }],
|
|
|
|
|
// What the Source observes across the repository it receives in full; elsewhere, only what addresses the agent.
|
|
|
|
|
scope: ({ meta }, config) => {
|
|
|
|
|
const row = scopeOf(meta)
|
|
|
|
|
return { listened: true, observed: row !== undefined && config[OBSERVED[row]], ...row ? { row } : {} }
|
|
|
|
|
},
|
|
|
|
|
health: ['gitea'],
|
|
|
|
|
channel: { send: sendGitea, link: giteaLink },
|
|
|
|
|
threadContext: giteaThreadContext,
|
|
|
|
|
// `#12` or `12`, or the page of an Issue or PR of any repository, names Thread 12.
|
|
|
|
|
find: query => {
|
|
|
|
|
const number = (/^#?(\d+)$/.exec(query.trim()) ?? /\/(?:issues|pulls)\/(\d+)/.exec(query))?.[1]
|
|
|
|
|
return { threads: number ? [`${SOURCE}:${number}`] : [], terms: [] }
|
|
|
|
|
},
|
|
|
|
|
policy: {
|
|
|
|
|
schema: z.object({ allowed: z.array(z.object({ id: z.string().regex(/^[A-Za-z0-9._-]+$/) }).strict()).default([]),
|
|
|
|
|
bots: z.array(z.object({ id: z.string().regex(/^[A-Za-z0-9._-]+$/) }).strict()).optional() }).strict(),
|
|
|
|
|
fields: [{ key: 'allowed', type: 'rows', label: { zh: '能发起工作的 Gitea 账号', en: 'Gitea accounts that may originate work' },
|
|
|
|
|
columns: [{ key: 'id', type: 'text', label: { zh: '账号', en: 'Account' }, pattern: '[A-Za-z0-9._\-]+' }] },
|
|
|
|
|
{ key: 'bots', type: 'rows', label: { zh: '评论可以成为候选的机器人', en: 'Bots whose comments may become candidates' },
|
|
|
|
|
hint: { zh: 'Gitea 的 API 不标记机器人账号，请把自动化账号列在这里：名单里的账号的消息在来源收到时成为推断的候选，交给评估。',
|
|
|
|
|
en: 'The Gitea API has no bot flag; list automation accounts here. A message of a bot on this list becomes an inferred candidate for an assessment when the Source receives it.' },
|
|
|
|
|
columns: [{ key: 'id', type: 'text', label: { zh: '账号', en: 'Account' }, pattern: '[A-Za-z0-9._\-]+' }] }],
|
|
|
|
|
explicit: { zh: '名单里的账号在 Issue 或评论里 @ 智能体的账号。', en: 'An account on the list mentions the agent\'s account in an Issue or a comment.' },
|
|
|
|
|
rows(stored) {
|
|
|
|
|
const config = GiteaConfigSchema.partial().parse(stored)
|
|
|
|
|
const note = (observed: boolean | undefined) => observed ? { zh: '全部', en: 'All' } : { zh: '仅 @', en: '@ only' }
|
|
|
|
|
const row = (id: Scope, label: { zh: string; en: string }) => ({ id, label, note: note(config[OBSERVED[id]]),
|
|
|
|
|
cells: { listedUnaddressed: !!config[OBSERVED[id]], unlistedAddressed: true, unlistedUnaddressed: !!config[OBSERVED[id]] } })
|
|
|
|
|
return [row('issues', { zh: '新 Issue', en: 'New Issues' }), row('pulls', { zh: '新 PR', en: 'New pull requests' }),
|
|
|
|
|
row('comments', { zh: '评论', en: 'Comments' })]
|
|
|
|
|
},
|
|
|
|
|
},
|
|
|
|
|
environment: {
|
|
|
|
|
variables: [
|
|
|
|
|
{ key: 'DSH_SUPERAGENT_GITEA_REPOSITORY', description: { zh: '这个来源监听的仓库，owner/name', en: 'The repository this Source watches, as owner/name' } },
|
|
|
|
|
{ key: 'DSH_SUPERAGENT_GITEA_URL', description: { zh: 'Gitea 实例地址，不带末尾斜杠', en: 'The Gitea instance URL without a trailing slash' } },
|
|
|
|
|
{ key: 'DSH_SUPERAGENT_GITEA_ACCOUNT', description: { zh: '智能体的 Gitea 账号；这里的 Issue 评论都以它的名义发出', en: 'The agent\'s Gitea account; Issue comments here go out under its name' } },
|
|
|
|
|
{ key: 'DSH_SUPERAGENT_GITEA_NUMBER', description: { zh: '接下工单的 Issue 或 PR 编号；工单不是从这个来源来的时为空', en: 'Number of the Issue or PR that started the ticket; empty when the ticket did not start on this Source' } },
|
|
|
|
|
],
|
|
|
|
|
values: ({ deployment, config, trigger }) => {
|
|
|
|
|
const number = trigger?.threads[0]?.source === SOURCE ? Number(trigger.meta.number) || 0 : 0
|
|
|
|
|
return { DSH_SUPERAGENT_GITEA_REPOSITORY: deployment.repository, DSH_SUPERAGENT_GITEA_URL: config.url || deployment.gitea?.url || '',
|
|
|
|
|
DSH_SUPERAGENT_GITEA_ACCOUNT: deployment.gitea?.login ?? '', DSH_SUPERAGENT_GITEA_NUMBER: number ? String(number) : '' }
|
|
|
|
|
},
|
|
|
|
|
},
|
|
|
|
|
facts: (deployment): Record<string, { zh: string; en: string }> => deployment?.gitea ? {
|
|
|
|
|
tokenEnv: { zh: `当前账号 @${deployment.gitea.login}`, en: `Current account @${deployment.gitea.login}` },
|
|
|
|
|
} : {},
|
|
|
|
|
}
|
|
|
|
|
export const name = 'dsh-superagent-gitea'
|
|
|
|
|
export const inject = ['orchestrator']
|
|
|
|
|
export function apply(ctx: Context): void {
|
|
|
|
|
ctx.effect(() => ctx.orchestrator.sources.register(giteaKind), 'gitea: source kind')
|
|
|
|
|
ctx.effect(() => {
|
|
|
|
|
const source = sourceContext(ctx.orchestrator, giteaKind)
|
|
|
|
|
const controller = new AbortController()
|
|
|
|
|
// Unhandled rejections terminate the DSH process, so a failure escaping the polling loop is reported as source health instead.
|
|
|
|
|
const running = watch(source, controller.signal)
|
|
|
|
|
.catch((error: unknown) => { source.report('gitea', { state: 'error', detail: String(error) }) })
|
|
|
|
|
return async () => { controller.abort(); await running }
|
|
|
|
|
})
|
|
|
|
|
}
|
|
|
|
|
async function watch(ctx: SourceContext, signal: AbortSignal): Promise<void> {
|
|
|
|
|
let interval = 60
|
|
|
|
|
let failures = 0
|
|
|
|
|
await ctx.ready
|
|
|
|
|
// Setup restarts dsh-superagent once it resolves the accounts; until then nothing is polled.
|
|
|
|
|
if (!ctx.configured()) { ctx.report('gitea', health('paused', 'setupPending')); return }
|
|
|
|
|
while (!signal.aborted) {
|
|
|
|
|
const config = ctx.config<GiteaConfig>(SOURCE)
|
|
|
|
|
try {
|
|
|
|
|
if (config) interval = await pollRound(ctx, signal)
|
|
|
|
|
else ctx.report('gitea', health('paused', 'sourceOff'))
|
|
|
|
|
failures = 0
|
|
|
|
|
} catch (error) {
|
|
|
|
|
failures += 1
|
|
|
|
|
const failed = pollFailure(failures, config?.failuresBeforeError ?? 1, error)
|
|
|
|
|
if (!signal.aborted && failed) ctx.report('gitea', failed)
|
|
|
|
|
}
|
|
|
|
|
await pause(Math.max(interval, ctx.config<GiteaConfig>(SOURCE)?.pollSeconds ?? 60) * 1000, signal)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
interface Round {
|
|
|
|
|
started: string; signal?: AbortSignal | undefined
|
|
|
|
|
read: Set<number>; failures: string[]; dropped: number[]; abandoned: number[]; tally: Tally
|
|
|
|
|
}
|
|
|
|
|
const newRound = (signal?: AbortSignal): Round => ({ started: new Date().toISOString(), signal, read: new Set(),
|
|
|
|
|
failures: [], dropped: [], abandoned: [], tally: { skipped: 0 } })
|
|
|
|
|
const stopped = (ctx: SourceContext, round: Round): boolean => !!round.signal?.aborted || !ctx.config(SOURCE)
|
|
|
|
|
const RETRY_WINDOW_MS = 24 * 3600_000
|
|
|
|
|
/** Gitea answers 404 for deleted Issues, PRs and comments; lost access fails the notification read and surfaces as health instead. */
|
|
|
|
|
const deleted = (error: unknown): boolean => error instanceof GiteaApiError && (error.status === 404 || error.status === 410)
|
|
|
|
|
/** The one API client of a round; every Gitea read and reply goes through the agent's own account. */
|
|
|
|
|
function api(ctx: SourceContext): GiteaApi {
|
|
|
|
|
const gitea = ctx.deployment.gitea
|
|
|
|
|
if (!gitea) throw new Error('The Gitea Source needs a Gitea deployment')
|
|
|
|
|
return giteaClient(gitea.url, gitea.tokenEnv)
|
|
|
|
|
}
|
|
|
|
|
const agentLogin = (ctx: SourceContext): string => ctx.deployment.gitea?.login ?? ''
|
|
|
|
|
/** Records a failed read on the Thread and in the round; deleted (404) or abandoned (24h+) threads do not fail the round. */
|
|
|
|
|
function failed(ctx: SourceContext, round: Round, number: number, error: unknown, abandoned = false): void {
|
|
|
|
|
if (deleted(error)) round.dropped.push(number)
|
|
|
|
|
else if (abandoned) round.abandoned.push(number)
|
|
|
|
|
else round.failures.push(`#${number}: ${String(error)}`)
|
|
|
|
|
ctx.threads.patch(`${SOURCE}:${number}`, { pollError: String(error) })
|
|
|
|
|
}
|
|
|
|
|
/**
|
|
|
|
|
* One polling round: notifications, then observed repository streams, then subscribed Threads. Each step and each Issue,
|
|
|
|
|
* PR or Thread fails independently so all steps run before the round fails.
|
|
|
|
|
* @returns the polling interval requested by the instance, in seconds.
|
|
|
|
|
*/
|
|
|
|
|
export async function pollRound(ctx: SourceContext, signal?: AbortSignal): Promise<number> {
|
|
|
|
|
const round = newRound(signal)
|
|
|
|
|
const notified = await pollNotifications(ctx, round).then(value => ({ value }), (error: unknown) => ({ error }))
|
|
|
|
|
await pollRepository(ctx, round)
|
|
|
|
|
await pollThreads(ctx, signal, round)
|
|
|
|
|
if ('error' in notified) throw notified.error
|
|
|
|
|
const { count, interval } = notified.value
|
|
|
|
|
if (signal?.aborted) return interval
|
|
|
|
|
if (round.failures.length) throw new Error(`${round.failures.length} Gitea reads failed: ${round.failures.slice(0, 3).join('; ')}`)
|
|
|
|
|
const named = (numbers: number[]) => [...new Set(numbers)].sort((a, b) => a - b).map(number => `#${number}`).join(', ')
|
|
|
|
|
const dropped = (round.dropped.length ? `; deleted on Gitea: ${named(round.dropped)}` : '')
|
|
|
|
|
+ (round.abandoned.length ? `; dropped after their reads kept failing: ${named(round.abandoned)}` : '')
|
|
|
|
|
+ (round.tally.skipped ? `; ${round.tally.skipped} unreadable items skipped` : '')
|
|
|
|
|
ctx.report('gitea', { state: 'ready', detail: `${count} notifications; next poll ≥ ${interval}s${dropped}`, at: round.started })
|
|
|
|
|
return interval
|
|
|
|
|
}
|
|
|
|
|
/** Reads notifications since the cursor and each Issue or PR they name, advancing the cursor atomically with the retry map. */
|
|
|
|
|
async function pollNotifications(ctx: SourceContext, round: Round): Promise<{ count: number; interval: number }> {
|
|
|
|
|
const { state } = ctx
|
|
|
|
|
const since = state.get(KV.gitea.since) ?? round.started
|
|
|
|
|
if (state.get(KV.gitea.since) === undefined) state.set(KV.gitea.since, since)
|
|
|
|
|
const { repository } = ctx.deployment
|
|
|
|
|
const response = await api(ctx).raw(`notifications?all=true&limit=50&since=${encodeURIComponent(since)}`)
|
|
|
|
|
const text = await response.text()
|
|
|
|
|
if (!response.ok) throw new GiteaApiError(response.status, `Gitea answered ${response.status} for notifications`, text.slice(0, 300))
|
|
|
|
|
const interval = Math.max(60, Number(response.headers.get('x-poll-interval') ?? 60) || 60)
|
|
|
|
|
const notifications = validItems(NotificationSchema, z.array(z.unknown()).parse(JSON.parse(text)), round.tally)
|
|
|
|
|
const retries = { ...state.get(KV.gitea.retries) }
|
|
|
|
|
// Failed items of the previous round run first, then newly notified items in chronological order; each Issue is read once.
|
|
|
|
|
const items = new Map(Object.entries(retries).map(([number, retry]) => [Number(number), retry]))
|
|
|
|
|
for (const notification of notifications.reverse()) {
|
|
|
|
|
if (!sameName(notification.repository.full_name, repository)) continue
|
|
|
|
|
const address = giteaAddress(notification.subject.url ?? '', repository)
|
|
|
|
|
if (!address || items.has(address.number)) continue
|
|
|
|
|
items.set(address.number, { since: [state.get(KV.gitea.cursor(address.number)) ?? since, since].sort().at(-1)!, pr: address.pr })
|
|
|
|
|
}
|
|
|
|
|
for (const [number, item] of items) {
|
|
|
|
|
if (stopped(ctx, round)) return { count: notifications.length, interval }
|
|
|
|
|
round.read.add(number)
|
|
|
|
|
try {
|
|
|
|
|
await readNotified(ctx, round, number, item)
|
|
|
|
|
delete retries[number]
|
|
|
|
|
} catch (error) {
|
|
|
|
|
if (stopped(ctx, round)) return { count: notifications.length, interval }
|
|
|
|
|
const retry = { ...item, failedAt: item.failedAt ?? round.started, attempts: (item.attempts ?? 0) + 1 }
|
|
|
|
|
// A permanently unreadable thread must not hold Source health in error forever; its cursor stays for later reads.
|
|
|
|
|
const abandoned = Date.parse(round.started) - Date.parse(retry.failedAt) >= RETRY_WINDOW_MS
|
|
|
|
|
failed(ctx, round, number, error, abandoned)
|
|
|
|
|
if (!deleted(error) && !abandoned) { retries[number] = retry; continue }
|
|
|
|
|
delete retries[number]
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
ctx.atomic(() => { state.set(KV.gitea.since, cursorAt(round.started)); state.set(KV.gitea.retries, retries) })
|
|
|
|
|
return { count: notifications.length, interval }
|
|
|
|
|
}
|
|
|
|
|
/** Reads one notified Issue or PR: its body when opened since the cursor, then its comments and, for a PR, its reviews. */
|
|
|
|
|
async function readNotified(ctx: SourceContext, round: Round, number: number, item: Retry): Promise<void> {
|
|
|
|
|
const { since } = item
|
|
|
|
|
const login = agentLogin(ctx)
|
|
|
|
|
const issue = await api(ctx).json(GiteaIssueSchema, `repos/${ctx.deployment.repository}/issues/${number}`)
|
|
|
|
|
const isPr = !!issue.pull_request || item.pr
|
|
|
|
|
if (Date.parse(issue.created_at) >= Date.parse(since) && !sameName(issue.user.login, login)) await ctx.receive(issueTrigger(issue, login))
|
|
|
|
|
const ref: GiteaRef = { number, isPr, url: issue.html_url, ...(isPr ? { prAuthor: issue.user.login } : {}) }
|
|
|
|
|
await intakeComments(ctx, round, ref, since)
|
|
|
|
|
if (isPr) await intakeReviews(ctx, round, ref, since)
|
|
|
|
|
nameThread(ctx, number, issue.title, issue.html_url)
|
|
|
|
|
ctx.state.set(KV.gitea.cursor(number), cursorAt(round.started))
|
|
|
|
|
ctx.threads.patch(`${SOURCE}:${number}`, { pollError: undefined })
|
|
|
|
|
}
|
|
|
|
|
async function intakeComments(ctx: SourceContext, round: Round, ref: GiteaRef, since: string): Promise<void> {
|
|
|
|
|
const path = `repos/${ctx.deployment.repository}/issues/${ref.number}/comments`
|
|
|
|
|
const comments = validItems(GiteaCommentSchema, await api(ctx).list(path, round.signal), round.tally)
|
|
|
|
|
const login = agentLogin(ctx)
|
|
|
|
|
for (const comment of comments) {
|
|
|
|
|
// Stopping mid-read throws so the thread's cursor stays where it was.
|
|
|
|
|
if (stopped(ctx, round)) throw new Error('The Gitea Source stopped during the poll')
|
|
|
|
|
const at = Date.parse(comment.updated_at ?? comment.created_at ?? '')
|
|
|
|
|
if (!Number.isFinite(at) || at < Date.parse(since)) continue
|
|
|
|
|
if (sameName(comment.user.login, login)) continue
|
|
|
|
|
const trigger = commentTrigger(comment, ref, login)
|
|
|
|
|
if (trigger.text) await ctx.receive(trigger)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
async function intakeReviews(ctx: SourceContext, round: Round, ref: GiteaRef, since: string): Promise<void> {
|
|
|
|
|
const path = `repos/${ctx.deployment.repository}/pulls/${ref.number}/reviews`
|
|
|
|
|
const reviews = validItems(GiteaReviewSchema, await api(ctx).list(path, round.signal), round.tally)
|
|
|
|
|
const login = agentLogin(ctx)
|
|
|
|
|
for (const review of reviews) {
|
|
|
|
|
if (stopped(ctx, round)) throw new Error('The Gitea Source stopped during the poll')
|
|
|
|
|
const at = Date.parse(review.updated_at ?? review.submitted_at ?? '')
|
|
|
|
|
if (!Number.isFinite(at) || at < Date.parse(since)) continue
|
|
|
|
|
if (sameName(review.user.login, login)) continue
|
|
|
|
|
const trigger = reviewTrigger(review, ref, login)
|
|
|
|
|
if (trigger.text) await ctx.receive(trigger)
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
/** One repository-wide stream; the stream's own schema parsed its items, so the branch casts to the union member it reads. */
|
|
|
|
|
interface Stream { id: Scope; observed: (config: GiteaConfig) => boolean; schema: z.ZodType<GiteaIssue | GiteaComment>; path: (repository: string) => string }
|
|
|
|
|
const STREAMS: Stream[] = [
|
|
|
|
|
{ id: 'issues', observed: config => config.observeIssues, schema: GiteaIssueSchema,
|
|
|
|
|
path: repository => `repos/${repository}/issues?state=all` },
|
|
|
|
|
{ id: 'pulls', observed: config => config.observePulls, schema: GiteaIssueSchema,
|
|
|
|
|
path: repository => `repos/${repository}/pulls?state=all` },
|
|
|
|
|
{ id: 'comments', observed: config => config.observeComments, schema: GiteaCommentSchema,
|
|
|
|
|
path: repository => `repos/${repository}/issues/comments` },
|
|
|
|
|
]
|
|
|
|
|
/** Polls the repository-wide streams (new Issues, new PRs, comments); each reads from when switched on, resets when off, and fails independently. */
|
|
|
|
|
async function pollRepository(ctx: SourceContext, round: Round): Promise<void> {
|
|
|
|
|
const config = ctx.config<GiteaConfig>(SOURCE)
|
|
|
|
|
if (!config) return
|
|
|
|
|
for (const stream of STREAMS) {
|
|
|
|
|
if (stopped(ctx, round)) return
|
|
|
|
|
if (!stream.observed(config)) { ctx.state.set(KV.gitea.stream(stream.id), undefined); continue }
|
|
|
|
|
try { await readStream(ctx, round, stream) }
|
|
|
|
|
catch (error) { if (!stopped(ctx, round)) round.failures.push(`${stream.id}: ${String(error)}`) }
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
/**
|
|
|
|
|
* Polls one repository-wide stream since its cursor and steps the cursor to five seconds before the newest item
|
|
|
|
|
* (`cursorAt`), relying on Trigger ID deduplication so repeated reads are no-ops. Items from before the kind was
|
|
|
|
|
* switched on or beyond the horizon are ignored.
|
|
|
|
|
*/
|
|
|
|
|
async function readStream(ctx: SourceContext, round: Round, stream: Stream): Promise<void> {
|
|
|
|
|
const name = KV.gitea.stream(stream.id)
|
|
|
|
|
const stored = ctx.state.get(name)
|
|
|
|
|
const started: Record<string, string> = { [stream.id]: stored?.started[stream.id] ?? round.started }
|
|
|
|
|
const horizon = Math.floor((Date.parse(round.started) - POLL_HORIZON_MS) / 3_600_000) * 3_600_000
|
|
|
|
|
const since = new Date(Math.max(Date.parse(stored?.since ?? round.started), horizon)).toISOString()
|
|
|
|
|
const items = validItems(stream.schema, await api(ctx).list(stream.path(ctx.deployment.repository), round.signal), round.tally)
|
|
|
|
|
const login = agentLogin(ctx)
|
|
|
|
|
const fresh = (at: string | null | undefined) =>
|
|
|
|
|
Date.parse(at ?? '') >= Math.max(Date.parse(started[stream.id]!), horizon) && Date.parse(at ?? '') >= Date.parse(since)
|
|
|
|
|
for (const item of items) {
|
|
|
|
|
// Stopping mid-stream aborts before advancing the cursor so unread items are retried.
|
|
|
|
|
if (stopped(ctx, round)) throw new Error('The Gitea Source stopped during the poll')
|
|
|
|
|
if (sameName(item.user.login, login)) continue
|
|
|
|
|
if (stream.id === 'comments') {
|
|
|
|
|
// The stream's own schema parsed every item, so the branch decides the member of the union.
|
|
|
|
|
const comment = item as GiteaComment
|
|
|
|
|
const number = commentNumber(comment.issue_url ?? '', ctx.deployment.repository)
|
|
|
|
|
if (!number || !fresh(comment.updated_at ?? comment.created_at)) continue
|
|
|
|
|
const ref: GiteaRef = { number, isPr: !!comment.pull_request_url, ...(comment.html_url ? { url: comment.html_url } : {}) }
|
|
|
|
|
const trigger = commentTrigger(comment, ref, login)
|
|
|
|
|
if (trigger.text) await ctx.receive(trigger)
|
|
|
|
|
continue
|
|
|
|
|
}
|
|
|
|
|
const issue = item as GiteaIssue
|
|
|
|
|
if (!fresh(issue.created_at)) continue
|
|
|
|
|
// `/pulls` items carry no `pull_request` meta; the stream itself names the kind, so pass it to the Trigger builder.
|
|
|
|
|
await ctx.receive(issueTrigger(issue, login, stream.id === 'pulls'))
|
|
|
|
|
nameThread(ctx, issue.number, issue.title, issue.html_url)
|
|
|
|
|
}
|
|
|
|
|
const newest = Math.max(...items.map(item => Date.parse(item.updated_at ?? '')).filter(Number.isFinite))
|
|
|
|
|
const moved = Number.isFinite(newest) ? cursorAt(new Date(newest).toISOString()) : since
|
|
|
|
|
const next = Date.parse(moved) > Date.parse(since) ? moved : since
|
|
|
|
|
ctx.state.set(name, { since: next, started })
|
|
|
|
|
}
|
|
|
|
|
/** Polls subscribed Issue and PR Threads not already read in this round. */
|
|
|
|
|
export async function pollThreads(ctx: SourceContext, signal?: AbortSignal, round = newRound(signal)): Promise<void> {
|
|
|
|
|
await pollEach(ctx, SOURCE, () => stopped(ctx, round), async thread => {
|
|
|
|
|
try { await pollThread(ctx, round, thread) }
|
|
|
|
|
catch (error) {
|
|
|
|
|
if (stopped(ctx, round)) throw error
|
|
|
|
|
if (deleted(error)) round.dropped.push(Number(thread.ref.id)); else round.failures.push(`${thread.key}: ${String(error)}`)
|
|
|
|
|
throw error
|
|
|
|
|
}
|
|
|
|
|
}, thread => round.read.has(Number(thread.ref.id)))
|
|
|
|
|
}
|
|
|
|
|
async function pollThread(ctx: SourceContext, round: Round, thread: Thread): Promise<void> {
|
|
|
|
|
const number = Number(thread.ref.id)
|
|
|
|
|
if (!number) throw new Error('Subscribed thread has no Gitea Issue or PR number')
|
|
|
|
|
round.read.add(number)
|
|
|
|
|
let isPr = thread.meta.isPr
|
|
|
|
|
if (typeof isPr !== 'boolean' || typeof thread.meta.title !== 'string') {
|
|
|
|
|
const issue = await api(ctx).json(GiteaIssueSchema, `repos/${ctx.deployment.repository}/issues/${number}`)
|
|
|
|
|
isPr = !!issue.pull_request
|
|
|
|
|
ctx.threads.patch(thread.key, { number, isPr, url: issue.html_url })
|
|
|
|
|
if (issue.title) nameThread(ctx, number, issue.title, issue.html_url)
|
|
|
|
|
}
|
|
|
|
|
const since = ctx.state.get(KV.gitea.cursor(number)) ?? thread.createdAt
|
|
|
|
|
const ref: GiteaRef = { number, isPr: isPr === true, ...(typeof thread.meta.url === 'string' ? { url: thread.meta.url } : {}) }
|
|
|
|
|
await intakeComments(ctx, round, ref, since)
|
|
|
|
|
if (isPr === true) await intakeReviews(ctx, round, ref, since)
|
|
|
|
|
ctx.state.set(KV.gitea.cursor(number), cursorAt(round.started))
|
|
|
|
|
}
|
|
|
|
|
function nameThread(ctx: SourceContext, number: number, title: string, url?: string): void {
|
|
|
|
|
ctx.threads.patch(`${SOURCE}:${number}`, { title })
|
|
|
|
|
ctx.threads.describe(`${SOURCE}:${number}`, { label: `#${number} ${title}`, ...(url ? { url } : {}),
|
|
|
|
|
facts: { number: { value: `#${number}` } } })
|
|
|
|
|
}
|
dev
