wehub-resource-sync
d25d482dc2
Publish CLI Package / publish-npm (push) Waiting to run
Publish Python SDK / publish-pypi (push) Waiting to run
Publish TypeScript SDK / publish-npm (push) Waiting to run
CI / Migrate Dev DB (push) Has been skipped
CI / Detect Version (push) Has been cancelled
CI / Migrate DB (push) Has been cancelled
CI / Build Dev ECR (./docker/app.Dockerfile, ECR_APP) (push) Has been cancelled
CI / Build Dev ECR (./docker/db.Dockerfile, ECR_MIGRATIONS) (push) Has been cancelled
CI / Build Dev ECR (./docker/pii.Dockerfile, ECR_PII) (push) Has been cancelled
CI / Build Dev ECR (./docker/realtime.Dockerfile, ECR_REALTIME) (push) Has been cancelled
CI / Deploy Trigger.dev (Dev) (push) Has been cancelled
CI / Build AMD64 (./docker/app.Dockerfile, ECR_APP, ghcr.io/simstudioai/simstudio) (push) Has been cancelled
CI / Build AMD64 (./docker/db.Dockerfile, ECR_MIGRATIONS, ghcr.io/simstudioai/migrations) (push) Has been cancelled
CI / Build AMD64 (./docker/pii.Dockerfile, ECR_PII, ghcr.io/simstudioai/pii) (push) Has been cancelled
CI / Build AMD64 (./docker/realtime.Dockerfile, ECR_REALTIME, ghcr.io/simstudioai/realtime) (push) Has been cancelled
CI / Build ARM64 (GHCR Only) (./docker/app.Dockerfile, ghcr.io/simstudioai/simstudio) (push) Has been cancelled
CI / Build ARM64 (GHCR Only) (./docker/db.Dockerfile, ghcr.io/simstudioai/migrations) (push) Has been cancelled
CI / Build ARM64 (GHCR Only) (./docker/pii.Dockerfile, ghcr.io/simstudioai/pii) (push) Has been cancelled
CI / Build ARM64 (GHCR Only) (./docker/realtime.Dockerfile, ghcr.io/simstudioai/realtime) (push) Has been cancelled
CI / Create GHCR Manifests (ghcr.io/simstudioai/migrations) (push) Has been cancelled
CI / Create GHCR Manifests (ghcr.io/simstudioai/pii) (push) Has been cancelled
CI / Create GHCR Manifests (ghcr.io/simstudioai/realtime) (push) Has been cancelled
CI / Create GHCR Manifests (ghcr.io/simstudioai/simstudio) (push) Has been cancelled
CI / Check Docs Changes (push) Has been cancelled
CI / Process Docs (push) Has been cancelled
CI / Create GitHub Release (push) Has been cancelled
CI / Test and Build (push) Has been cancelled
1715 lines
58 KiB
TypeScript
1715 lines
58 KiB
TypeScript
import { createLogger } from '@sim/logger'
|
|
import { getErrorMessage, toError } from '@sim/utils/errors'
|
|
import { fetchWithRetry, VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/documents/utils'
|
|
import { azureDevopsConnectorMeta } from '@/connectors/azure-devops/meta'
|
|
import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types'
|
|
import {
|
|
CONNECTOR_MAX_FILE_BYTES,
|
|
htmlToPlainText,
|
|
joinTagArray,
|
|
markSkipped,
|
|
parseTagDate,
|
|
readBodyWithLimit,
|
|
sizeLimitSkipReason,
|
|
} from '@/connectors/utils'
|
|
|
|
const logger = createLogger('AzureDevOpsConnector')
|
|
|
|
const ADO_BASE_URL = 'https://dev.azure.com'
|
|
const WIKI_API_VERSION = '7.1'
|
|
const WIKIS_LIST_API_VERSION = '7.1'
|
|
const WIQL_API_VERSION = '7.1'
|
|
const WORKITEMS_API_VERSION = '7.1'
|
|
const PROJECT_API_VERSION = '7.1'
|
|
const GIT_API_VERSION = '7.1'
|
|
|
|
/** Page size for the wiki `pagesbatch` endpoint. */
|
|
const WIKI_PAGE_BATCH_SIZE = 100
|
|
/** Page size for the WIQL → workitemsbatch listing pipeline. ADO caps a batch at 200 ids. */
|
|
const WORK_ITEM_BATCH_SIZE = 200
|
|
/** Concurrency for per-page wiki ETag lookups during listing. */
|
|
const WIKI_ETAG_CONCURRENCY = 5
|
|
/** Page size for paginating repository-file stubs out of the in-memory tree. */
|
|
const FILE_BATCH_SIZE = 100
|
|
/**
|
|
* Max repository file size to index. The Items list API does not return file
|
|
* size, so this cap is enforced at content-fetch time in getDocument: the raw
|
|
* octet-stream body is read through `readBodyWithLimit`, which streams the bytes
|
|
* and aborts (returning null) the moment the cap is exceeded. Larger files are
|
|
* skipped without being fully buffered.
|
|
*/
|
|
const MAX_FILE_SIZE = CONNECTOR_MAX_FILE_BYTES
|
|
/** Bytes sniffed for a NUL byte when detecting binary files (matches git's heuristic). */
|
|
const BINARY_SNIFF_BYTES = 8000
|
|
/**
|
|
* WIQL returns at most 20,000 work item references. We cap `$top` at this bound
|
|
* so the connector never silently relies on truncated results; users who need
|
|
* more should narrow the query via the work-item filters.
|
|
*/
|
|
const WIQL_MAX_RESULTS = 20000
|
|
|
|
/**
|
|
* externalId discriminators. Wiki pages are addressed by `wiki:{wikiId}:{path}`,
|
|
* work items by `wi:{id}`, and repository files by `file:{repoId}:{path}`.
|
|
*/
|
|
const FILE_PREFIX = 'file:'
|
|
|
|
type ContentType = 'wiki' | 'workitems' | 'files' | 'both' | 'all'
|
|
|
|
/** Listing phases, walked in order: wiki ➜ work items ➜ repository files. */
|
|
type SyncPhase = 'wiki' | 'workitems' | 'file'
|
|
|
|
/**
|
|
* Returns the ordered list of active sync phases for a content-type choice.
|
|
* Phase order is fixed (wiki ➜ workitems ➜ file) so the phase-encoded cursor and
|
|
* the maxItems phase-boundary guard compose deterministically.
|
|
*/
|
|
function activePhases(contentType: ContentType): SyncPhase[] {
|
|
const phases: SyncPhase[] = []
|
|
if (contentType === 'wiki' || contentType === 'both' || contentType === 'all') phases.push('wiki')
|
|
if (contentType === 'workitems' || contentType === 'both' || contentType === 'all') {
|
|
phases.push('workitems')
|
|
}
|
|
if (contentType === 'files' || contentType === 'all') phases.push('file')
|
|
return phases
|
|
}
|
|
|
|
/**
|
|
* Returns the phase following `current` for a content type, or undefined when
|
|
* `current` is the last active phase.
|
|
*/
|
|
function nextPhase(current: SyncPhase, contentType: ContentType): SyncPhase | undefined {
|
|
const phases = activePhases(contentType)
|
|
const idx = phases.indexOf(current)
|
|
return idx >= 0 && idx + 1 < phases.length ? phases[idx + 1] : undefined
|
|
}
|
|
|
|
/**
|
|
* Builds the Azure DevOps PAT auth header. ADO PATs authenticate via HTTP Basic
|
|
* with an empty username and the token as the password.
|
|
*/
|
|
function patAuthHeader(accessToken: string): string {
|
|
return `Basic ${Buffer.from(`:${accessToken}`).toString('base64')}`
|
|
}
|
|
|
|
/**
|
|
* Normalizes the configured content type, defaulting to wiki pages.
|
|
*/
|
|
function parseContentType(value: unknown): ContentType {
|
|
if (value === 'workitems' || value === 'files' || value === 'both' || value === 'all') {
|
|
return value
|
|
}
|
|
return 'wiki'
|
|
}
|
|
|
|
/**
|
|
* Heuristic binary detection: a NUL byte in the first 8 KB marks the file as
|
|
* binary, matching `git diff` / `git grep` semantics.
|
|
*/
|
|
function isBinaryBuffer(buf: Buffer): boolean {
|
|
const len = Math.min(buf.length, BINARY_SNIFF_BYTES)
|
|
for (let i = 0; i < len; i++) {
|
|
if (buf[i] === 0) return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
/**
|
|
* Parses a comma-separated extension filter into a normalized set (leading dot,
|
|
* lowercased). Returns null when no filter is configured (accept all files).
|
|
*/
|
|
function parseExtensions(raw: unknown): Set<string> | null {
|
|
const trimmed = typeof raw === 'string' ? raw.trim() : ''
|
|
if (!trimmed) return null
|
|
const exts = trimmed
|
|
.split(',')
|
|
.map((e) => e.trim().toLowerCase())
|
|
.filter(Boolean)
|
|
.map((e) => (e.startsWith('.') ? e : `.${e}`))
|
|
return exts.length > 0 ? new Set(exts) : null
|
|
}
|
|
|
|
/**
|
|
* Returns true when the file path matches the extension filter (or no filter set).
|
|
*/
|
|
function matchesExtension(filePath: string, extSet: Set<string> | null): boolean {
|
|
if (!extSet) return true
|
|
const lastDot = filePath.lastIndexOf('.')
|
|
if (lastDot === -1) return false
|
|
return extSet.has(filePath.slice(lastDot).toLowerCase())
|
|
}
|
|
|
|
/**
|
|
* Strips the `refs/heads/` prefix from a default-branch ref so it can be used as
|
|
* a `versionDescriptor.version` branch name.
|
|
*/
|
|
function stripRefsHeads(ref: string): string {
|
|
return ref.replace(/^refs\/heads\//, '')
|
|
}
|
|
|
|
/**
|
|
* Reads a trimmed string config value, returning '' when absent.
|
|
*/
|
|
function readString(value: unknown): string {
|
|
return typeof value === 'string' ? value.trim() : ''
|
|
}
|
|
|
|
/**
|
|
* Escapes a value for safe interpolation into a single-quoted WIQL string literal.
|
|
* WIQL escapes an embedded single quote by doubling it.
|
|
*/
|
|
function escapeWiql(value: string): string {
|
|
return value.replace(/'/g, "''")
|
|
}
|
|
|
|
/**
|
|
* Encodes an external ID that combines a discriminator with its identifier,
|
|
* e.g. `wiki:{wikiId}:{pagePath}` or `wi:{id}`.
|
|
*/
|
|
function workItemExternalId(id: number): string {
|
|
return `wi:${id}`
|
|
}
|
|
|
|
function wikiPageExternalId(wikiId: string, pagePath: string): string {
|
|
return `wiki:${wikiId}:${pagePath}`
|
|
}
|
|
|
|
/**
|
|
* Parses a wiki external ID back into its wiki ID and page path.
|
|
*/
|
|
function parseWikiExternalId(externalId: string): { wikiId: string; pagePath: string } | null {
|
|
if (!externalId.startsWith('wiki:')) return null
|
|
const rest = externalId.slice('wiki:'.length)
|
|
const sep = rest.indexOf(':')
|
|
if (sep === -1) return null
|
|
return { wikiId: rest.slice(0, sep), pagePath: rest.slice(sep + 1) }
|
|
}
|
|
|
|
/**
|
|
* Builds the externalId for a repository file: `file:{repoId}:{path}`. The path
|
|
* retains its leading slash as returned by the Items API.
|
|
*/
|
|
function fileExternalId(repoId: string, path: string): string {
|
|
return `${FILE_PREFIX}${repoId}:${path}`
|
|
}
|
|
|
|
/**
|
|
* Parses a file externalId back into its repository ID and path. Returns null
|
|
* when the externalId is not a file ID.
|
|
*/
|
|
function parseFileExternalId(externalId: string): { repoId: string; path: string } | null {
|
|
if (!externalId.startsWith(FILE_PREFIX)) return null
|
|
const rest = externalId.slice(FILE_PREFIX.length)
|
|
const sep = rest.indexOf(':')
|
|
if (sep === -1) return null
|
|
return { repoId: rest.slice(0, sep), path: rest.slice(sep + 1) }
|
|
}
|
|
|
|
/**
|
|
* Builds the change-detection hash for a repository file. The git blob objectId
|
|
* is content-addressable, so it changes exactly when the file content changes,
|
|
* and it is reported both by the tree listing (`objectId`) and the per-file
|
|
* metadata fetch (`objectId`) — so the listing stub and the hydrated document
|
|
* normally hash identically without a content fetch during listing.
|
|
*
|
|
* Hydration in getFileDocument is a two-step fetch against the same branch ref:
|
|
* a JSON metadata call yields the objectId used for this hash, then a raw
|
|
* octet-stream call yields the content. Both pin to the branch *name*, not a
|
|
* commit SHA, so if the branch advances between the two calls the stored hash
|
|
* (metadata call's objectId) and the stored content (content call's bytes) can
|
|
* be one commit apart. This window is bounded and self-heals: the next listing
|
|
* reports the branch's current objectId, which differs from the stored
|
|
* one-commit-old hash, queuing an update that re-fetches and re-converges
|
|
* content and hash. (A revert to identical bytes yields the identical objectId
|
|
* by content-addressing, so the stored content is already correct in that case.)
|
|
*/
|
|
function buildFileContentHash(repoId: string, objectId: string): string {
|
|
return `ado:file:${repoId}:${objectId}`
|
|
}
|
|
|
|
interface WikiV2 {
|
|
id: string
|
|
name: string
|
|
remoteUrl?: string
|
|
type?: string
|
|
}
|
|
|
|
interface GitRepository {
|
|
id: string
|
|
name: string
|
|
defaultBranch?: string
|
|
isDisabled?: boolean
|
|
remoteUrl?: string
|
|
webUrl?: string
|
|
size?: number
|
|
}
|
|
|
|
interface GitItem {
|
|
objectId: string
|
|
gitObjectType?: string
|
|
path: string
|
|
isFolder?: boolean
|
|
content?: string
|
|
contentMetadata?: {
|
|
isBinary?: boolean
|
|
fileName?: string
|
|
encoding?: number
|
|
}
|
|
}
|
|
|
|
/**
|
|
* A repository file flattened across all in-scope repositories, carrying enough
|
|
* context to build its stub and source URL during offset-based pagination.
|
|
*/
|
|
interface RepoFileEntry {
|
|
repoId: string
|
|
repoName: string
|
|
repoWebUrl?: string
|
|
branch: string
|
|
item: GitItem
|
|
}
|
|
|
|
interface WikiPageDetail {
|
|
id: number
|
|
path: string
|
|
}
|
|
|
|
interface WorkItemRef {
|
|
id: number
|
|
}
|
|
|
|
interface RawWorkItem {
|
|
id: number
|
|
rev?: number
|
|
url?: string
|
|
fields?: Record<string, unknown>
|
|
}
|
|
|
|
/**
|
|
* Resolves the change-detection revision for a work item. ADO returns the
|
|
* revision as the top-level `rev` property on each batch item; `System.Rev` is
|
|
* not guaranteed to be echoed in the requested `fields`, so `rev` is the
|
|
* authoritative source. Falls back to the in-fields rev, then `System.ChangedDate`.
|
|
*/
|
|
function resolveWorkItemRev(raw: RawWorkItem, fields: Record<string, unknown>): string {
|
|
if (typeof raw.rev === 'number') return String(raw.rev)
|
|
const fieldRev = fields['System.Rev']
|
|
if (typeof fieldRev === 'number') return String(fieldRev)
|
|
const changed = fields['System.ChangedDate']
|
|
if (typeof changed === 'string' && changed) return changed
|
|
return '0'
|
|
}
|
|
|
|
/**
|
|
* Fetches the list of wikis in the configured project. Returns an empty list on
|
|
* 401/403/404 so a missing or inaccessible wiki feature degrades gracefully
|
|
* rather than aborting the sync.
|
|
*/
|
|
async function listWikis(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
retryOptions?: Parameters<typeof fetchWithRetry>[2],
|
|
syncContext?: Record<string, unknown>
|
|
): Promise<WikiV2[]> {
|
|
const url = `${ADO_BASE_URL}/${encodeURIComponent(organization)}/${encodeURIComponent(project)}/_apis/wiki/wikis?api-version=${WIKIS_LIST_API_VERSION}`
|
|
const response = await fetchWithRetry(
|
|
url,
|
|
{
|
|
method: 'GET',
|
|
headers: { Accept: 'application/json', Authorization: patAuthHeader(accessToken) },
|
|
},
|
|
retryOptions
|
|
)
|
|
if (!response.ok) {
|
|
if (response.status === 401 || response.status === 403 || response.status === 404) {
|
|
/**
|
|
* 401/403 mean the wikis still exist but this PAT cannot read them right
|
|
* now — flag the listing as incomplete so reconciliation does not delete
|
|
* previously synced wiki pages. A 404 means the wiki feature/content is
|
|
* genuinely absent, so reconciliation stays enabled.
|
|
*/
|
|
if ((response.status === 401 || response.status === 403) && syncContext) {
|
|
syncContext.listingCapped = true
|
|
}
|
|
logger.warn('Azure DevOps wikis unavailable; skipping wiki listing', {
|
|
organization,
|
|
project,
|
|
status: response.status,
|
|
})
|
|
return []
|
|
}
|
|
const errorText = await response.text().catch(() => '')
|
|
logger.error('Failed to list Azure DevOps wikis', { status: response.status, error: errorText })
|
|
throw new Error(`Failed to list wikis: ${response.status}`)
|
|
}
|
|
const data = await response.json()
|
|
return (data.value as WikiV2[] | undefined) ?? []
|
|
}
|
|
|
|
/**
|
|
* Resolves the wikis for the project, caching them on the sync context so a
|
|
* single sync (and its deferred getDocument calls) reuse one listing.
|
|
*/
|
|
async function resolveWikis(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
syncContext?: Record<string, unknown>
|
|
): Promise<WikiV2[]> {
|
|
const cached = syncContext?.wikis as WikiV2[] | undefined
|
|
if (cached) return cached
|
|
const wikis = await listWikis(accessToken, organization, project, undefined, syncContext)
|
|
if (syncContext) syncContext.wikis = wikis
|
|
return wikis
|
|
}
|
|
|
|
/**
|
|
* Returns true when the wiki should be included given an optional wiki filter
|
|
* (matched case-insensitively against the wiki id or name).
|
|
*/
|
|
function wikiMatchesFilter(wiki: WikiV2, filter: string): boolean {
|
|
if (!filter) return true
|
|
const needle = filter.toLowerCase()
|
|
return wiki.id.toLowerCase() === needle || (wiki.name ?? '').toLowerCase() === needle
|
|
}
|
|
|
|
/**
|
|
* Fetches the ETag for a single wiki page without downloading its content.
|
|
* The ETag changes whenever the page is edited, making it a reliable
|
|
* metadata-only change-detection hash for the deferred-content pattern.
|
|
*/
|
|
async function fetchWikiPageETag(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
wikiId: string,
|
|
pagePath: string
|
|
): Promise<string | null> {
|
|
const url = `${ADO_BASE_URL}/${encodeURIComponent(organization)}/${encodeURIComponent(project)}/_apis/wiki/wikis/${encodeURIComponent(wikiId)}/pages?path=${encodeURIComponent(pagePath)}&api-version=${WIKI_API_VERSION}`
|
|
const response = await fetchWithRetry(url, {
|
|
method: 'GET',
|
|
headers: { Accept: 'application/json', Authorization: patAuthHeader(accessToken) },
|
|
})
|
|
if (!response.ok) {
|
|
if (response.status === 404) return null
|
|
logger.warn('Failed to fetch wiki page ETag', { pagePath, status: response.status })
|
|
return null
|
|
}
|
|
const etag = response.headers.get('etag')
|
|
return etag ? etag.replace(/"/g, '') : null
|
|
}
|
|
|
|
/**
|
|
* Builds a wiki page stub. The contentHash is derived from the page ETag
|
|
* (falls back to the page id when no ETag is available), guaranteeing the
|
|
* hash is identical between listing and content fetch.
|
|
*/
|
|
function wikiPageToStub(
|
|
organization: string,
|
|
project: string,
|
|
wiki: WikiV2,
|
|
page: WikiPageDetail,
|
|
etag: string | null
|
|
): ExternalDocument {
|
|
const title = page.path.split('/').filter(Boolean).pop() || page.path || 'Untitled'
|
|
const sourceUrl = wiki.remoteUrl
|
|
? `${wiki.remoteUrl}?pagePath=${encodeURIComponent(page.path)}`
|
|
: undefined
|
|
return {
|
|
externalId: wikiPageExternalId(wiki.id, page.path),
|
|
title,
|
|
content: '',
|
|
contentDeferred: true,
|
|
mimeType: 'text/plain',
|
|
sourceUrl,
|
|
contentHash: `ado:wiki:${wiki.id}:${page.path}:${etag ?? page.id}`,
|
|
metadata: {
|
|
kind: 'wiki',
|
|
organization,
|
|
project,
|
|
wikiId: wiki.id,
|
|
wikiName: wiki.name,
|
|
pageId: page.id,
|
|
pagePath: page.path,
|
|
},
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Builds a work item document. Work items are returned inline (not deferred)
|
|
* because the batch fetch already includes all field content. The contentHash
|
|
* uses the work item revision, which increments on every change. HTML-bearing
|
|
* fields (description, repro steps, acceptance criteria) are stripped to text.
|
|
*/
|
|
function workItemToDocument(
|
|
organization: string,
|
|
project: string,
|
|
raw: RawWorkItem
|
|
): ExternalDocument {
|
|
const fields = raw.fields ?? {}
|
|
const title = (fields['System.Title'] as string | undefined) ?? `Work Item ${raw.id}`
|
|
const workItemType = (fields['System.WorkItemType'] as string | undefined) ?? ''
|
|
const state = (fields['System.State'] as string | undefined) ?? ''
|
|
const rev = resolveWorkItemRev(raw, fields)
|
|
const changedDate = (fields['System.ChangedDate'] as string | undefined) ?? ''
|
|
const areaPath = (fields['System.AreaPath'] as string | undefined) ?? ''
|
|
const iterationPath = (fields['System.IterationPath'] as string | undefined) ?? ''
|
|
const rawTags = (fields['System.Tags'] as string | undefined) ?? ''
|
|
const tags = rawTags
|
|
.split(';')
|
|
.map((t) => t.trim())
|
|
.filter(Boolean)
|
|
const description = htmlToPlainText((fields['System.Description'] as string | undefined) ?? '')
|
|
const reproSteps = htmlToPlainText(
|
|
(fields['Microsoft.VSTS.TCM.ReproSteps'] as string | undefined) ?? ''
|
|
)
|
|
const acceptanceCriteria = htmlToPlainText(
|
|
(fields['Microsoft.VSTS.Common.AcceptanceCriteria'] as string | undefined) ?? ''
|
|
)
|
|
|
|
const contentParts: string[] = [`Title: ${title}`, `Type: ${workItemType}`, `State: ${state}`]
|
|
if (tags.length > 0) contentParts.push(`Tags: ${tags.join(', ')}`)
|
|
if (description) contentParts.push('', 'Description:', description)
|
|
if (reproSteps) contentParts.push('', 'Repro Steps:', reproSteps)
|
|
if (acceptanceCriteria) contentParts.push('', 'Acceptance Criteria:', acceptanceCriteria)
|
|
|
|
return {
|
|
externalId: workItemExternalId(raw.id),
|
|
title: `#${raw.id}: ${title}`,
|
|
content: contentParts.join('\n'),
|
|
contentDeferred: false,
|
|
mimeType: 'text/plain',
|
|
sourceUrl: `${ADO_BASE_URL}/${encodeURIComponent(organization)}/${encodeURIComponent(project)}/_workitems/edit/${raw.id}`,
|
|
contentHash: `ado:wi:${raw.id}:${rev}`,
|
|
metadata: {
|
|
kind: 'workitem',
|
|
organization,
|
|
project,
|
|
workItemId: raw.id,
|
|
workItemType,
|
|
state,
|
|
areaPath,
|
|
iterationPath,
|
|
tags,
|
|
changedDate,
|
|
rev,
|
|
},
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Reads the work-item filter configuration from sourceConfig.
|
|
*/
|
|
interface WorkItemFilters {
|
|
workItemType: string
|
|
state: string
|
|
areaPath: string
|
|
tags: string[]
|
|
customWiql: string
|
|
}
|
|
|
|
function readWorkItemFilters(sourceConfig: Record<string, unknown>): WorkItemFilters {
|
|
const tagsRaw = readString(sourceConfig.workItemTags)
|
|
const tags = tagsRaw
|
|
? tagsRaw
|
|
.split(',')
|
|
.map((t) => t.trim())
|
|
.filter(Boolean)
|
|
: []
|
|
return {
|
|
workItemType: readString(sourceConfig.workItemType),
|
|
state: readString(sourceConfig.state),
|
|
areaPath: readString(sourceConfig.areaPath),
|
|
tags,
|
|
customWiql: readString(sourceConfig.customWiql),
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Builds the WIQL query for the configured work-item filters. User-supplied
|
|
* values are escaped against WIQL string-literal injection. `lastSyncAt`
|
|
* narrows results to items changed since the previous sync, and `idAfter`
|
|
* restricts to items with a greater id (used to probe past the 20,000-item
|
|
* WIQL cap).
|
|
*
|
|
* A custom WIQL query is used verbatim: neither the incremental changed-date
|
|
* filter nor the probe condition can be injected into arbitrary user WIQL
|
|
* safely, so custom queries always run as full listings on every sync. Change
|
|
* detection still short-circuits unchanged items via the content hash.
|
|
*/
|
|
function buildWiql(filters: WorkItemFilters, lastSyncAt?: Date, idAfter?: number): string {
|
|
if (filters.customWiql) return filters.customWiql
|
|
|
|
const clauses: string[] = ['[System.TeamProject] = @project']
|
|
if (filters.workItemType) {
|
|
clauses.push(`[System.WorkItemType] = '${escapeWiql(filters.workItemType)}'`)
|
|
}
|
|
if (filters.state) {
|
|
clauses.push(`[System.State] = '${escapeWiql(filters.state)}'`)
|
|
}
|
|
if (filters.areaPath) {
|
|
clauses.push(`[System.AreaPath] UNDER '${escapeWiql(filters.areaPath)}'`)
|
|
}
|
|
for (const tag of filters.tags) {
|
|
clauses.push(`[System.Tags] CONTAINS '${escapeWiql(tag)}'`)
|
|
}
|
|
if (lastSyncAt) {
|
|
clauses.push(`[System.ChangedDate] >= '${lastSyncAt.toISOString()}'`)
|
|
}
|
|
if (idAfter !== undefined) {
|
|
clauses.push(`[System.Id] > ${idAfter}`)
|
|
}
|
|
|
|
return `SELECT [System.Id] FROM workitems WHERE ${clauses.join(' AND ')} ORDER BY [System.ChangedDate] DESC`
|
|
}
|
|
|
|
/**
|
|
* Runs a WIQL query for work items in the project and returns their IDs.
|
|
* WIQL itself is not paginated and returns at most 20,000 ids; pagination
|
|
* happens over the resulting ID list via the workitemsbatch endpoint.
|
|
*/
|
|
async function queryWorkItemIds(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
wiql: string,
|
|
top: number = WIQL_MAX_RESULTS
|
|
): Promise<number[]> {
|
|
const url = `${ADO_BASE_URL}/${encodeURIComponent(organization)}/${encodeURIComponent(project)}/_apis/wit/wiql?$top=${top}&api-version=${WIQL_API_VERSION}`
|
|
const response = await fetchWithRetry(url, {
|
|
method: 'POST',
|
|
headers: {
|
|
Accept: 'application/json',
|
|
'Content-Type': 'application/json',
|
|
Authorization: patAuthHeader(accessToken),
|
|
},
|
|
body: JSON.stringify({ query: wiql }),
|
|
})
|
|
if (!response.ok) {
|
|
const errorText = await response.text().catch(() => '')
|
|
logger.error('Failed to query Azure DevOps work items', {
|
|
status: response.status,
|
|
error: errorText,
|
|
})
|
|
throw new Error(`Failed to query work items: ${response.status}`)
|
|
}
|
|
const data = await response.json()
|
|
const refs = (data.workItems as WorkItemRef[] | undefined) ?? []
|
|
if (refs.length >= WIQL_MAX_RESULTS) {
|
|
logger.warn('WIQL result hit the 20,000-item cap; narrow work-item filters to sync all items', {
|
|
organization,
|
|
project,
|
|
})
|
|
}
|
|
return refs.map((ref) => ref.id)
|
|
}
|
|
|
|
/**
|
|
* Fetches full field details for a batch of work item IDs (max 200 per call).
|
|
* `errorPolicy: 'Omit'` keeps the batch resilient: a single inaccessible or
|
|
* deleted id is dropped from the response rather than failing the whole call.
|
|
*/
|
|
async function fetchWorkItemsBatch(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
ids: number[]
|
|
): Promise<RawWorkItem[]> {
|
|
if (ids.length === 0) return []
|
|
const url = `${ADO_BASE_URL}/${encodeURIComponent(organization)}/${encodeURIComponent(project)}/_apis/wit/workitemsbatch?api-version=${WORKITEMS_API_VERSION}`
|
|
const response = await fetchWithRetry(url, {
|
|
method: 'POST',
|
|
headers: {
|
|
Accept: 'application/json',
|
|
'Content-Type': 'application/json',
|
|
Authorization: patAuthHeader(accessToken),
|
|
},
|
|
body: JSON.stringify({
|
|
ids,
|
|
errorPolicy: 'Omit',
|
|
fields: [
|
|
'System.Id',
|
|
'System.Title',
|
|
'System.WorkItemType',
|
|
'System.State',
|
|
'System.AreaPath',
|
|
'System.IterationPath',
|
|
'System.ChangedDate',
|
|
'System.Tags',
|
|
'System.Description',
|
|
'Microsoft.VSTS.TCM.ReproSteps',
|
|
'Microsoft.VSTS.Common.AcceptanceCriteria',
|
|
],
|
|
}),
|
|
})
|
|
if (!response.ok) {
|
|
const errorText = await response.text().catch(() => '')
|
|
logger.error('Failed to fetch Azure DevOps work items batch', {
|
|
status: response.status,
|
|
error: errorText,
|
|
})
|
|
throw new Error(`Failed to fetch work items batch: ${response.status}`)
|
|
}
|
|
const data = await response.json()
|
|
return (data.value as RawWorkItem[] | undefined) ?? []
|
|
}
|
|
|
|
/**
|
|
* Reads the repository-file filter configuration from sourceConfig.
|
|
*/
|
|
interface FileFilters {
|
|
repositoryName: string
|
|
branch: string
|
|
pathPrefix: string
|
|
extensions: Set<string> | null
|
|
}
|
|
|
|
function readFileFilters(sourceConfig: Record<string, unknown>): FileFilters {
|
|
const rawPrefix = readString(sourceConfig.pathPrefix)
|
|
return {
|
|
repositoryName: readString(sourceConfig.repositoryName),
|
|
branch: readString(sourceConfig.branch),
|
|
pathPrefix: rawPrefix,
|
|
extensions: parseExtensions(sourceConfig.fileExtensions),
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Lists the project's git repositories. Returns an empty list on 401/403/404 so
|
|
* a project without Git or without repo access degrades gracefully instead of
|
|
* aborting the sync.
|
|
*/
|
|
async function listRepositories(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
retryOptions?: Parameters<typeof fetchWithRetry>[2],
|
|
syncContext?: Record<string, unknown>
|
|
): Promise<GitRepository[]> {
|
|
const url = `${ADO_BASE_URL}/${encodeURIComponent(organization)}/${encodeURIComponent(project)}/_apis/git/repositories?api-version=${GIT_API_VERSION}`
|
|
const response = await fetchWithRetry(
|
|
url,
|
|
{
|
|
method: 'GET',
|
|
headers: { Accept: 'application/json', Authorization: patAuthHeader(accessToken) },
|
|
},
|
|
retryOptions
|
|
)
|
|
if (!response.ok) {
|
|
if (response.status === 401 || response.status === 403 || response.status === 404) {
|
|
/**
|
|
* 401/403 mean repositories still exist but this PAT cannot read them
|
|
* right now — flag the listing as incomplete so reconciliation does not
|
|
* delete previously synced repository files. A 404 means the Git feature
|
|
* is genuinely absent, so reconciliation stays enabled.
|
|
*/
|
|
if ((response.status === 401 || response.status === 403) && syncContext) {
|
|
syncContext.listingCapped = true
|
|
}
|
|
logger.warn('Azure DevOps repositories unavailable; skipping file listing', {
|
|
organization,
|
|
project,
|
|
status: response.status,
|
|
})
|
|
return []
|
|
}
|
|
const errorText = await response.text().catch(() => '')
|
|
logger.error('Failed to list Azure DevOps repositories', {
|
|
status: response.status,
|
|
error: errorText,
|
|
})
|
|
throw new Error(`Failed to list repositories: ${response.status}`)
|
|
}
|
|
const data = await response.json()
|
|
return (data.value as GitRepository[] | undefined) ?? []
|
|
}
|
|
|
|
/**
|
|
* Resolves the in-scope repositories for the project, caching them on the sync
|
|
* context so a single sync reuses one listing. Disabled repositories and, when a
|
|
* filter is set, non-matching repositories are excluded.
|
|
*/
|
|
async function resolveRepositories(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
repositoryFilter: string,
|
|
syncContext?: Record<string, unknown>
|
|
): Promise<GitRepository[]> {
|
|
const cached = syncContext?.repositories as GitRepository[] | undefined
|
|
const all =
|
|
cached ?? (await listRepositories(accessToken, organization, project, undefined, syncContext))
|
|
if (syncContext && !cached) syncContext.repositories = all
|
|
|
|
const needle = repositoryFilter.toLowerCase()
|
|
return all.filter((repo) => {
|
|
if (repo.isDisabled) return false
|
|
if (!needle) return true
|
|
return repo.id.toLowerCase() === needle || (repo.name ?? '').toLowerCase() === needle
|
|
})
|
|
}
|
|
|
|
/**
|
|
* Lists every blob in a repository at the given branch via the non-paginated
|
|
* Items list API (recursionLevel=Full). Returns an empty list on 401/403/404 so
|
|
* a single inaccessible or empty repo does not abort the sync.
|
|
*/
|
|
async function listRepositoryBlobs(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
repoId: string,
|
|
branch: string,
|
|
syncContext?: Record<string, unknown>
|
|
): Promise<GitItem[]> {
|
|
const params = new URLSearchParams({
|
|
recursionLevel: 'Full',
|
|
'versionDescriptor.version': branch,
|
|
'versionDescriptor.versionType': 'Branch',
|
|
'api-version': GIT_API_VERSION,
|
|
})
|
|
const url = `${ADO_BASE_URL}/${encodeURIComponent(organization)}/${encodeURIComponent(project)}/_apis/git/repositories/${encodeURIComponent(repoId)}/items?${params.toString()}`
|
|
const response = await fetchWithRetry(url, {
|
|
method: 'GET',
|
|
headers: { Accept: 'application/json', Authorization: patAuthHeader(accessToken) },
|
|
})
|
|
if (!response.ok) {
|
|
if (response.status === 401 || response.status === 403 || response.status === 404) {
|
|
/**
|
|
* 401/403 mean the repository's files still exist but this PAT cannot
|
|
* read them right now — flag the listing as incomplete so reconciliation
|
|
* does not delete previously synced files. A 404 means the branch/repo
|
|
* content is genuinely absent (empty repo, deleted branch), so
|
|
* reconciliation stays enabled.
|
|
*/
|
|
if ((response.status === 401 || response.status === 403) && syncContext) {
|
|
syncContext.listingCapped = true
|
|
}
|
|
logger.warn('Azure DevOps repository items unavailable; skipping repository', {
|
|
repoId,
|
|
branch,
|
|
status: response.status,
|
|
})
|
|
return []
|
|
}
|
|
const errorText = await response.text().catch(() => '')
|
|
logger.error('Failed to list Azure DevOps repository items', {
|
|
repoId,
|
|
branch,
|
|
status: response.status,
|
|
error: errorText,
|
|
})
|
|
throw new Error(`Failed to list repository items: ${response.status}`)
|
|
}
|
|
/**
|
|
* The Items list API documents no pagination, but very large trees may emit
|
|
* an `x-ms-continuationtoken` response header. No request parameter exists
|
|
* to follow it, so when it appears the tree is treated as incomplete: the
|
|
* listing is flagged so deletion reconciliation cannot remove files that
|
|
* were never returned.
|
|
*/
|
|
if (response.headers.get('x-ms-continuationtoken')) {
|
|
if (syncContext) syncContext.listingCapped = true
|
|
logger.warn(
|
|
'Azure DevOps repository tree listing returned a continuation token; partial tree',
|
|
{
|
|
repoId,
|
|
branch,
|
|
}
|
|
)
|
|
}
|
|
const data = await response.json()
|
|
const items = (data.value as GitItem[] | undefined) ?? []
|
|
return items.filter((item) => item.gitObjectType === 'blob' && !item.isFolder && item.path)
|
|
}
|
|
|
|
/**
|
|
* Builds the web UI URL for a repository file at a given branch. Azure DevOps
|
|
* file links use `{repoWebUrl}?path={path}&version=GB{branch}` (GB = Git Branch).
|
|
*/
|
|
function buildFileSourceUrl(
|
|
repoWebUrl: string | undefined,
|
|
branch: string,
|
|
path: string
|
|
): string | undefined {
|
|
if (!repoWebUrl) return undefined
|
|
return `${repoWebUrl}?path=${encodeURIComponent(path)}&version=GB${encodeURIComponent(branch)}`
|
|
}
|
|
|
|
/**
|
|
* Builds a deferred stub for a repository file. Content is empty and fetched
|
|
* lazily via getDocument for new/changed files only. The contentHash is the git
|
|
* blob objectId, identical between the stub and the hydrated document.
|
|
*/
|
|
function fileToStub(organization: string, project: string, entry: RepoFileEntry): ExternalDocument {
|
|
const path = entry.item.path
|
|
const title = path.split('/').filter(Boolean).pop() || path
|
|
return {
|
|
externalId: fileExternalId(entry.repoId, path),
|
|
title,
|
|
content: '',
|
|
contentDeferred: true,
|
|
mimeType: 'text/plain',
|
|
sourceUrl: buildFileSourceUrl(entry.repoWebUrl, entry.branch, path),
|
|
contentHash: buildFileContentHash(entry.repoId, entry.item.objectId),
|
|
metadata: {
|
|
kind: 'file',
|
|
organization,
|
|
project,
|
|
repository: entry.repoName,
|
|
repositoryId: entry.repoId,
|
|
branch: entry.branch,
|
|
path,
|
|
},
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Resolves the flattened, filtered list of repository files for the configured
|
|
* scope. Repositories are listed once, each is walked via the recursive Items
|
|
* API, and blobs are filtered by path prefix and extension. The result is cached
|
|
* on syncContext so offset-based pagination and the maxItems cap apply over a
|
|
* stable list across pages.
|
|
*/
|
|
async function resolveRepoFiles(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
filters: FileFilters,
|
|
syncContext?: Record<string, unknown>
|
|
): Promise<RepoFileEntry[]> {
|
|
const cached = syncContext?.repoFiles as RepoFileEntry[] | undefined
|
|
if (cached) return cached
|
|
|
|
const repositories = await resolveRepositories(
|
|
accessToken,
|
|
organization,
|
|
project,
|
|
filters.repositoryName,
|
|
syncContext
|
|
)
|
|
|
|
const normalizedPrefix =
|
|
filters.pathPrefix && !filters.pathPrefix.startsWith('/')
|
|
? `/${filters.pathPrefix}`
|
|
: filters.pathPrefix
|
|
|
|
const entries: RepoFileEntry[] = []
|
|
for (const repo of repositories) {
|
|
const branch = filters.branch || stripRefsHeads(repo.defaultBranch ?? '')
|
|
if (!branch) {
|
|
/**
|
|
* No branch override and no resolvable default branch. An empty
|
|
* repository (size 0) has nothing to list and nothing previously synced,
|
|
* so it is skipped without flagging — flagging here would permanently
|
|
* suppress deletion reconciliation for any project containing an empty
|
|
* repo. A non-empty repository reaching this branch means content exists
|
|
* but its default branch ref is missing/unreadable, so the listing is
|
|
* flagged incomplete to protect previously synced files from
|
|
* reconciliation deletion.
|
|
*/
|
|
if ((repo.size ?? 0) > 0 && syncContext) {
|
|
syncContext.listingCapped = true
|
|
}
|
|
logger.warn('Skipping Azure DevOps repository with no default branch', {
|
|
repoId: repo.id,
|
|
repoName: repo.name,
|
|
size: repo.size ?? 0,
|
|
})
|
|
continue
|
|
}
|
|
const blobs = await listRepositoryBlobs(
|
|
accessToken,
|
|
organization,
|
|
project,
|
|
repo.id,
|
|
branch,
|
|
syncContext
|
|
)
|
|
for (const item of blobs) {
|
|
if (normalizedPrefix && !item.path.startsWith(normalizedPrefix)) continue
|
|
if (!matchesExtension(item.path, filters.extensions)) continue
|
|
entries.push({
|
|
repoId: repo.id,
|
|
repoName: repo.name,
|
|
repoWebUrl: repo.webUrl,
|
|
branch,
|
|
item,
|
|
})
|
|
}
|
|
}
|
|
|
|
if (syncContext) syncContext.repoFiles = entries
|
|
return entries
|
|
}
|
|
|
|
/**
|
|
* Lists a single batch of repository-file stubs. The full filtered file list is
|
|
* resolved once and cached on syncContext; the cursor is an offset into that
|
|
* list, of the form `file|{offset}`.
|
|
*/
|
|
async function listRepoFiles(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
filters: FileFilters,
|
|
maxItems: number,
|
|
cursor: string | undefined,
|
|
syncContext: Record<string, unknown> | undefined
|
|
): Promise<ExternalDocumentList> {
|
|
const entries = await resolveRepoFiles(accessToken, organization, project, filters, syncContext)
|
|
|
|
if (entries.length === 0) {
|
|
return { documents: [], hasMore: false }
|
|
}
|
|
|
|
let offset = 0
|
|
if (cursor) {
|
|
const parts = cursor.split('|')
|
|
offset = Number(parts[1]) || 0
|
|
}
|
|
|
|
const chunk = entries.slice(offset, offset + FILE_BATCH_SIZE)
|
|
const documents = chunk.map((entry) => fileToStub(organization, project, entry))
|
|
|
|
const nextOffset = offset + FILE_BATCH_SIZE
|
|
const { documents: capped, capped: hitLimit } = applyMaxItemsCap(
|
|
documents,
|
|
maxItems,
|
|
syncContext,
|
|
nextOffset < entries.length
|
|
)
|
|
|
|
const hasMore = !hitLimit && nextOffset < entries.length
|
|
|
|
return {
|
|
documents: capped,
|
|
nextCursor: hasMore ? `file|${nextOffset}` : undefined,
|
|
hasMore,
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Resolves the branch to fetch a single repository file from in getDocument. Uses
|
|
* the configured branch override when set, otherwise the repository's default
|
|
* branch (resolved from the cached or freshly-listed repository record).
|
|
*/
|
|
async function resolveFileBranch(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
repoId: string,
|
|
branchOverride: string,
|
|
syncContext?: Record<string, unknown>
|
|
): Promise<{ branch: string; repo?: GitRepository }> {
|
|
if (branchOverride) {
|
|
const repos = (syncContext?.repositories as GitRepository[] | undefined) ?? []
|
|
return { branch: branchOverride, repo: repos.find((r) => r.id === repoId) }
|
|
}
|
|
const repos =
|
|
(syncContext?.repositories as GitRepository[] | undefined) ??
|
|
(await listRepositories(accessToken, organization, project))
|
|
if (syncContext && !syncContext.repositories) syncContext.repositories = repos
|
|
const repo = repos.find((r) => r.id === repoId)
|
|
return { branch: stripRefsHeads(repo?.defaultBranch ?? ''), repo }
|
|
}
|
|
|
|
/**
|
|
* Fetches and hydrates a single repository file by its externalId. Re-fetches the
|
|
* item with content, rebuilds the objectId-based hash identically to the stub,
|
|
* and skips binary, oversized, or empty files. Returns null for 404 / not found.
|
|
*/
|
|
async function getFileDocument(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
externalId: string,
|
|
branchOverride: string,
|
|
syncContext?: Record<string, unknown>
|
|
): Promise<ExternalDocument | null> {
|
|
const parsed = parseFileExternalId(externalId)
|
|
if (!parsed) return null
|
|
const { repoId, path } = parsed
|
|
|
|
const { branch, repo } = await resolveFileBranch(
|
|
accessToken,
|
|
organization,
|
|
project,
|
|
repoId,
|
|
branchOverride,
|
|
syncContext
|
|
)
|
|
if (!branch) {
|
|
logger.warn('Cannot resolve branch for Azure DevOps file', { externalId })
|
|
return null
|
|
}
|
|
|
|
const metadataParams = new URLSearchParams({
|
|
path,
|
|
'versionDescriptor.version': branch,
|
|
'versionDescriptor.versionType': 'Branch',
|
|
includeContentMetadata: 'true',
|
|
$format: 'json',
|
|
'api-version': GIT_API_VERSION,
|
|
})
|
|
const metadataUrl = `${ADO_BASE_URL}/${encodeURIComponent(organization)}/${encodeURIComponent(project)}/_apis/git/repositories/${encodeURIComponent(repoId)}/items?${metadataParams.toString()}`
|
|
const metadataResponse = await fetchWithRetry(metadataUrl, {
|
|
method: 'GET',
|
|
headers: { Accept: 'application/json', Authorization: patAuthHeader(accessToken) },
|
|
})
|
|
|
|
if (!metadataResponse.ok) {
|
|
if (metadataResponse.status === 404) return null
|
|
throw new Error(`Failed to fetch repository file metadata: ${metadataResponse.status}`)
|
|
}
|
|
|
|
const item = (await metadataResponse.json()) as GitItem
|
|
if (!item.objectId) return null
|
|
if (item.contentMetadata?.isBinary) {
|
|
logger.info('Skipping binary Azure DevOps file', { path })
|
|
return null
|
|
}
|
|
|
|
/**
|
|
* Content is fetched as raw bytes (Accept: application/octet-stream) rather
|
|
* than via `includeContent=true` JSON. The JSON `content` field's encoding is
|
|
* ambiguous (the API may deliver base64 or codepage-transcoded text per
|
|
* `ItemContentType`), whereas the octet-stream response is the byte-exact git
|
|
* blob, which is then binary-sniffed and decoded as UTF-8.
|
|
*/
|
|
const contentParams = new URLSearchParams({
|
|
path,
|
|
'versionDescriptor.version': branch,
|
|
'versionDescriptor.versionType': 'Branch',
|
|
'api-version': GIT_API_VERSION,
|
|
})
|
|
const contentUrl = `${ADO_BASE_URL}/${encodeURIComponent(organization)}/${encodeURIComponent(project)}/_apis/git/repositories/${encodeURIComponent(repoId)}/items?${contentParams.toString()}`
|
|
const contentResponse = await fetchWithRetry(contentUrl, {
|
|
method: 'GET',
|
|
headers: { Accept: 'application/octet-stream', Authorization: patAuthHeader(accessToken) },
|
|
})
|
|
|
|
if (!contentResponse.ok) {
|
|
if (contentResponse.status === 404) return null
|
|
throw new Error(`Failed to fetch repository file content: ${contentResponse.status}`)
|
|
}
|
|
|
|
const buffer = await readBodyWithLimit(contentResponse, MAX_FILE_SIZE)
|
|
if (buffer === null) {
|
|
logger.info('Skipping oversized Azure DevOps file', { path })
|
|
const skippedTitle = path.split('/').filter(Boolean).pop() || path
|
|
return markSkipped(
|
|
{
|
|
externalId,
|
|
title: skippedTitle,
|
|
content: '',
|
|
mimeType: 'text/plain',
|
|
sourceUrl: buildFileSourceUrl(repo?.webUrl, branch, path),
|
|
contentHash: buildFileContentHash(repoId, item.objectId),
|
|
metadata: {
|
|
kind: 'file',
|
|
organization,
|
|
project,
|
|
repository: repo?.name ?? '',
|
|
repositoryId: repoId,
|
|
branch,
|
|
path,
|
|
},
|
|
},
|
|
sizeLimitSkipReason(MAX_FILE_SIZE)
|
|
)
|
|
}
|
|
if (isBinaryBuffer(buffer)) {
|
|
logger.info('Skipping binary Azure DevOps file', { path })
|
|
return null
|
|
}
|
|
|
|
const content = buffer.toString('utf8')
|
|
if (!content.trim()) return null
|
|
|
|
const title = path.split('/').filter(Boolean).pop() || path
|
|
return {
|
|
externalId,
|
|
title,
|
|
content,
|
|
contentDeferred: false,
|
|
mimeType: 'text/plain',
|
|
sourceUrl: buildFileSourceUrl(repo?.webUrl, branch, path),
|
|
contentHash: buildFileContentHash(repoId, item.objectId),
|
|
metadata: {
|
|
kind: 'file',
|
|
organization,
|
|
project,
|
|
repository: repo?.name ?? '',
|
|
repositoryId: repoId,
|
|
branch,
|
|
path,
|
|
size: buffer.byteLength,
|
|
},
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Applies the optional maxItems cap to a batch, tracking the running total in
|
|
* syncContext and flagging `listingCapped` when the cap actually truncated the
|
|
* listing. The sync engine reads `listingCapped` to suppress deletion
|
|
* reconciliation on a truncated listing — without it, a capped full sync would
|
|
* wrongly delete every source document beyond the cap.
|
|
*
|
|
* `moreAvailable` tells the helper whether the current phase has further items
|
|
* beyond this page. The flag is only set when documents were actually dropped
|
|
* (this page was sliced, or more pages exist) — when the cap merely coincides
|
|
* with source exhaustion, reconciliation stays enabled so deleted source
|
|
* documents are still cleaned up.
|
|
*/
|
|
function applyMaxItemsCap(
|
|
documents: ExternalDocument[],
|
|
maxItems: number,
|
|
syncContext: Record<string, unknown> | undefined,
|
|
moreAvailable: boolean
|
|
): { documents: ExternalDocument[]; capped: boolean } {
|
|
if (maxItems <= 0) return { documents, capped: false }
|
|
const prevTotal = (syncContext?.totalDocsFetched as number) ?? 0
|
|
const remaining = Math.max(0, maxItems - prevTotal)
|
|
const slicedSome = documents.length > remaining
|
|
const sliced = slicedSome ? documents.slice(0, remaining) : documents
|
|
const newTotal = prevTotal + sliced.length
|
|
if (syncContext) syncContext.totalDocsFetched = newTotal
|
|
const capped = newTotal >= maxItems
|
|
if (capped && (slicedSome || moreAvailable) && syncContext) syncContext.listingCapped = true
|
|
return { documents: sliced, capped }
|
|
}
|
|
|
|
/**
|
|
* Lists a single batch of wiki pages across the project's wikis (optionally
|
|
* filtered to one wiki). Uses a compound cursor of the form
|
|
* `wiki|{wikiIndex}|{continuationToken}` so each wiki's `pagesbatch` pagination
|
|
* is tracked independently.
|
|
*/
|
|
async function listWikiPages(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
wikiFilter: string,
|
|
maxItems: number,
|
|
cursor: string | undefined,
|
|
syncContext?: Record<string, unknown>
|
|
): Promise<ExternalDocumentList> {
|
|
const allWikis = await resolveWikis(accessToken, organization, project, syncContext)
|
|
const wikis = allWikis.filter((w) => wikiMatchesFilter(w, wikiFilter))
|
|
|
|
if (wikis.length === 0) {
|
|
return { documents: [], hasMore: false }
|
|
}
|
|
|
|
let wikiIndex = 0
|
|
let continuationToken: string | undefined
|
|
if (cursor) {
|
|
// The continuation token is opaque and may contain `|`; keep everything after
|
|
// the second separator intact instead of truncating it with a naive split.
|
|
const firstSep = cursor.indexOf('|')
|
|
const secondSep = firstSep === -1 ? -1 : cursor.indexOf('|', firstSep + 1)
|
|
if (secondSep !== -1) {
|
|
wikiIndex = Number(cursor.slice(firstSep + 1, secondSep)) || 0
|
|
const token = cursor.slice(secondSep + 1)
|
|
continuationToken = token || undefined
|
|
}
|
|
}
|
|
|
|
if (wikiIndex >= wikis.length) {
|
|
return { documents: [], hasMore: false }
|
|
}
|
|
|
|
const wiki = wikis[wikiIndex]
|
|
const url = `${ADO_BASE_URL}/${encodeURIComponent(organization)}/${encodeURIComponent(project)}/_apis/wiki/wikis/${encodeURIComponent(wiki.id)}/pagesbatch?api-version=${WIKI_API_VERSION}`
|
|
const body: Record<string, unknown> = { top: WIKI_PAGE_BATCH_SIZE }
|
|
if (continuationToken) body.continuationToken = continuationToken
|
|
|
|
const response = await fetchWithRetry(url, {
|
|
method: 'POST',
|
|
headers: {
|
|
Accept: 'application/json',
|
|
'Content-Type': 'application/json',
|
|
Authorization: patAuthHeader(accessToken),
|
|
},
|
|
body: JSON.stringify(body),
|
|
})
|
|
if (!response.ok) {
|
|
const errorText = await response.text().catch(() => '')
|
|
logger.error('Failed to list Azure DevOps wiki pages', {
|
|
wikiId: wiki.id,
|
|
status: response.status,
|
|
error: errorText,
|
|
})
|
|
throw new Error(`Failed to list wiki pages: ${response.status}`)
|
|
}
|
|
|
|
const data = await response.json()
|
|
const pages = (data.value as WikiPageDetail[] | undefined) ?? []
|
|
const nextContinuation = response.headers.get('x-ms-continuationtoken') || undefined
|
|
|
|
const documents: ExternalDocument[] = []
|
|
for (let i = 0; i < pages.length; i += WIKI_ETAG_CONCURRENCY) {
|
|
const batch = pages.slice(i, i + WIKI_ETAG_CONCURRENCY)
|
|
const stubs = await Promise.all(
|
|
batch.map(async (page) => {
|
|
const etag = await fetchWikiPageETag(accessToken, organization, project, wiki.id, page.path)
|
|
return wikiPageToStub(organization, project, wiki, page, etag)
|
|
})
|
|
)
|
|
documents.push(...stubs)
|
|
}
|
|
|
|
const { documents: capped, capped: hitLimit } = applyMaxItemsCap(
|
|
documents,
|
|
maxItems,
|
|
syncContext,
|
|
Boolean(nextContinuation) || wikiIndex + 1 < wikis.length
|
|
)
|
|
if (hitLimit) {
|
|
return { documents: capped, hasMore: false }
|
|
}
|
|
|
|
let nextCursor: string | undefined
|
|
let hasMore: boolean
|
|
if (nextContinuation) {
|
|
nextCursor = `wiki|${wikiIndex}|${nextContinuation}`
|
|
hasMore = true
|
|
} else if (wikiIndex + 1 < wikis.length) {
|
|
nextCursor = `wiki|${wikiIndex + 1}|`
|
|
hasMore = true
|
|
} else {
|
|
hasMore = false
|
|
}
|
|
|
|
return { documents: capped, nextCursor, hasMore }
|
|
}
|
|
|
|
/**
|
|
* Lists a single batch of work items. The full ID list is resolved once via WIQL
|
|
* and cached on the sync context; the cursor is an offset into that list.
|
|
*/
|
|
async function listWorkItems(
|
|
accessToken: string,
|
|
organization: string,
|
|
project: string,
|
|
filters: WorkItemFilters,
|
|
maxItems: number,
|
|
cursor: string | undefined,
|
|
syncContext: Record<string, unknown> | undefined,
|
|
lastSyncAt: Date | undefined
|
|
): Promise<ExternalDocumentList> {
|
|
let ids = syncContext?.workItemIds as number[] | undefined
|
|
if (!ids) {
|
|
const wiql = buildWiql(filters, lastSyncAt)
|
|
ids = await queryWorkItemIds(accessToken, organization, project, wiql)
|
|
if (syncContext) syncContext.workItemIds = ids
|
|
|
|
if (ids.length >= WIQL_MAX_RESULTS && syncContext) {
|
|
/**
|
|
* The WIQL result filled the 20,000-item cap. Distinguish an exact fit
|
|
* from genuine truncation: for structured filters, probe for any
|
|
* matching item with an id beyond the largest returned one and only
|
|
* flag the listing incomplete when one exists — otherwise deletion
|
|
* reconciliation would be disabled forever for a project with exactly
|
|
* 20,000 matching items. Custom WIQL cannot be probed (no safe clause
|
|
* injection), so it is flagged conservatively.
|
|
*/
|
|
let truncated = true
|
|
if (!filters.customWiql) {
|
|
let maxId = 0
|
|
for (const id of ids) {
|
|
if (id > maxId) maxId = id
|
|
}
|
|
const probeWiql = buildWiql(filters, lastSyncAt, maxId)
|
|
const beyond = await queryWorkItemIds(accessToken, organization, project, probeWiql, 1)
|
|
truncated = beyond.length > 0
|
|
}
|
|
if (truncated) {
|
|
syncContext.listingCapped = true
|
|
}
|
|
}
|
|
}
|
|
|
|
if (ids.length === 0) {
|
|
return { documents: [], hasMore: false }
|
|
}
|
|
|
|
let offset = 0
|
|
if (cursor) {
|
|
const parts = cursor.split('|')
|
|
offset = Number(parts[1]) || 0
|
|
}
|
|
|
|
const chunk = ids.slice(offset, offset + WORK_ITEM_BATCH_SIZE)
|
|
const raw = await fetchWorkItemsBatch(accessToken, organization, project, chunk)
|
|
if (raw.length < chunk.length && syncContext) {
|
|
syncContext.listingCapped = true
|
|
logger.warn(
|
|
'workitemsbatch omitted ids that WIQL returned; flagging listing as incomplete so reconciliation skips deletion',
|
|
{ requested: chunk.length, returned: raw.length, organization, project }
|
|
)
|
|
}
|
|
const documents = raw.map((item) => workItemToDocument(organization, project, item))
|
|
|
|
const nextOffset = offset + WORK_ITEM_BATCH_SIZE
|
|
const { documents: capped, capped: hitLimit } = applyMaxItemsCap(
|
|
documents,
|
|
maxItems,
|
|
syncContext,
|
|
nextOffset < ids.length
|
|
)
|
|
|
|
const hasMore = !hitLimit && nextOffset < ids.length
|
|
|
|
return {
|
|
documents: capped,
|
|
nextCursor: hasMore ? `wi|${nextOffset}` : undefined,
|
|
hasMore,
|
|
}
|
|
}
|
|
|
|
export const azureDevopsConnector: ConnectorConfig = {
|
|
...azureDevopsConnectorMeta,
|
|
|
|
listDocuments: async (
|
|
accessToken: string,
|
|
sourceConfig: Record<string, unknown>,
|
|
cursor?: string,
|
|
syncContext?: Record<string, unknown>,
|
|
lastSyncAt?: Date
|
|
): Promise<ExternalDocumentList> => {
|
|
const organization = readString(sourceConfig.organization)
|
|
const project = readString(sourceConfig.project)
|
|
const contentType = parseContentType(sourceConfig.contentType)
|
|
const wikiFilter = readString(sourceConfig.wikiName)
|
|
const filters = readWorkItemFilters(sourceConfig)
|
|
const fileFilters = readFileFilters(sourceConfig)
|
|
const maxItems = sourceConfig.maxItems ? Number(sourceConfig.maxItems) : 0
|
|
|
|
if (!organization || !project) {
|
|
throw new Error('Organization and project are required')
|
|
}
|
|
|
|
const phases = activePhases(contentType)
|
|
if (phases.length === 0) return { documents: [], hasMore: false }
|
|
|
|
/**
|
|
* Resolves which phase a cursor belongs to. Phases run in a fixed order
|
|
* (wiki ➜ workitems ➜ file) and each phase owns a cursor prefix
|
|
* (`wiki|`, `wi|`, `file|`). A missing cursor starts at the first active phase.
|
|
*/
|
|
const cursorPhase: SyncPhase = cursor?.startsWith('wi|')
|
|
? 'workitems'
|
|
: cursor?.startsWith('file|')
|
|
? 'file'
|
|
: 'wiki'
|
|
|
|
/**
|
|
* A cursor from a phase that is no longer active (e.g. the content-type
|
|
* config changed) is discarded along with its offsets — otherwise another
|
|
* phase would misparse its tokens as numeric offsets and skip documents.
|
|
*/
|
|
const cursorIsActive = phases.includes(cursorPhase)
|
|
const phase = cursorIsActive ? cursorPhase : phases[0]
|
|
const initialCursor = cursorIsActive ? cursor : undefined
|
|
|
|
/** Lists a single batch for the given phase. The cursor is passed only when it belongs to that phase. */
|
|
const runPhase = (target: SyncPhase, phaseCursor: string | undefined) => {
|
|
if (target === 'wiki') {
|
|
return listWikiPages(
|
|
accessToken,
|
|
organization,
|
|
project,
|
|
wikiFilter,
|
|
maxItems,
|
|
phaseCursor,
|
|
syncContext
|
|
)
|
|
}
|
|
if (target === 'workitems') {
|
|
return listWorkItems(
|
|
accessToken,
|
|
organization,
|
|
project,
|
|
filters,
|
|
maxItems,
|
|
phaseCursor,
|
|
syncContext,
|
|
lastSyncAt
|
|
)
|
|
}
|
|
return listRepoFiles(
|
|
accessToken,
|
|
organization,
|
|
project,
|
|
fileFilters,
|
|
maxItems,
|
|
phaseCursor,
|
|
syncContext
|
|
)
|
|
}
|
|
|
|
/** True once the maxItems cap has been reached during this sync run. */
|
|
const capReached = () =>
|
|
maxItems > 0 && ((syncContext?.totalDocsFetched as number) ?? 0) >= maxItems
|
|
|
|
/**
|
|
* Walks phases starting at `phase`, accumulating documents. Within a phase,
|
|
* pagination is driven by that phase's own cursor; when a phase is exhausted
|
|
* the walker advances to the next active phase (resetting its cursor). The
|
|
* maxItems cap is honored at phase boundaries so the cap is never exceeded
|
|
* across phases.
|
|
*/
|
|
let current: SyncPhase | undefined = phase
|
|
let phaseCursor = initialCursor
|
|
const documents: ExternalDocument[] = []
|
|
|
|
while (current) {
|
|
const result = await runPhase(current, phaseCursor)
|
|
documents.push(...result.documents)
|
|
|
|
if (result.hasMore) {
|
|
return { documents, nextCursor: result.nextCursor, hasMore: true }
|
|
}
|
|
if (capReached()) {
|
|
const remainingPhase = nextPhase(current, contentType)
|
|
if (remainingPhase && syncContext) {
|
|
syncContext.listingCapped = true
|
|
}
|
|
return { documents, hasMore: false }
|
|
}
|
|
current = nextPhase(current, contentType)
|
|
phaseCursor = undefined
|
|
}
|
|
|
|
return { documents, hasMore: false }
|
|
},
|
|
|
|
getDocument: async (
|
|
accessToken: string,
|
|
sourceConfig: Record<string, unknown>,
|
|
externalId: string,
|
|
syncContext?: Record<string, unknown>
|
|
): Promise<ExternalDocument | null> => {
|
|
const organization = readString(sourceConfig.organization)
|
|
const project = readString(sourceConfig.project)
|
|
if (!organization || !project) {
|
|
throw new Error('Organization and project are required')
|
|
}
|
|
|
|
/**
|
|
* Repository files are deferred and re-fetched here. Work items are returned
|
|
* inline during listing, so getDocument is otherwise only invoked for
|
|
* deferred wiki pages. Unknown IDs return null defensively.
|
|
*/
|
|
if (externalId.startsWith(FILE_PREFIX)) {
|
|
try {
|
|
return await getFileDocument(
|
|
accessToken,
|
|
organization,
|
|
project,
|
|
externalId,
|
|
readString(sourceConfig.branch),
|
|
syncContext
|
|
)
|
|
} catch (error) {
|
|
logger.warn(`Failed to fetch Azure DevOps file ${externalId}`, {
|
|
error: toError(error).message,
|
|
})
|
|
return null
|
|
}
|
|
}
|
|
|
|
const parsed = parseWikiExternalId(externalId)
|
|
if (!parsed) return null
|
|
|
|
const { wikiId, pagePath } = parsed
|
|
|
|
let wikiName: string | undefined
|
|
let remoteUrl: string | undefined
|
|
try {
|
|
const wikis = await resolveWikis(accessToken, organization, project, syncContext)
|
|
const wiki = wikis.find((w) => w.id === wikiId)
|
|
wikiName = wiki?.name
|
|
remoteUrl = wiki?.remoteUrl
|
|
} catch (error) {
|
|
logger.warn('Failed to resolve wiki metadata for page', {
|
|
externalId,
|
|
error: toError(error).message,
|
|
})
|
|
}
|
|
|
|
const url = `${ADO_BASE_URL}/${encodeURIComponent(organization)}/${encodeURIComponent(project)}/_apis/wiki/wikis/${encodeURIComponent(wikiId)}/pages?path=${encodeURIComponent(pagePath)}&includeContent=true&api-version=${WIKI_API_VERSION}`
|
|
const response = await fetchWithRetry(url, {
|
|
method: 'GET',
|
|
headers: { Accept: 'application/json', Authorization: patAuthHeader(accessToken) },
|
|
})
|
|
|
|
if (!response.ok) {
|
|
if (response.status === 404) return null
|
|
throw new Error(`Failed to fetch wiki page: ${response.status}`)
|
|
}
|
|
|
|
const etag = response.headers.get('etag')
|
|
const data = await response.json()
|
|
const content = (data.content as string | undefined) ?? ''
|
|
if (!content.trim()) return null
|
|
|
|
const pageId = (data.id as number | undefined) ?? 0
|
|
const title = pagePath.split('/').filter(Boolean).pop() || pagePath || 'Untitled'
|
|
const sourceUrl = remoteUrl
|
|
? `${remoteUrl}?pagePath=${encodeURIComponent(pagePath)}`
|
|
: ((data.remoteUrl as string | undefined) ?? undefined)
|
|
|
|
return {
|
|
externalId,
|
|
title,
|
|
content,
|
|
contentDeferred: false,
|
|
mimeType: 'text/plain',
|
|
sourceUrl,
|
|
contentHash: `ado:wiki:${wikiId}:${pagePath}:${etag ? etag.replace(/"/g, '') : pageId}`,
|
|
metadata: {
|
|
kind: 'wiki',
|
|
organization,
|
|
project,
|
|
wikiId,
|
|
wikiName,
|
|
pageId,
|
|
pagePath,
|
|
},
|
|
}
|
|
},
|
|
|
|
validateConfig: async (
|
|
accessToken: string,
|
|
sourceConfig: Record<string, unknown>
|
|
): Promise<{ valid: boolean; error?: string }> => {
|
|
const organization = readString(sourceConfig.organization)
|
|
const project = readString(sourceConfig.project)
|
|
|
|
if (!organization || !project) {
|
|
return { valid: false, error: 'Organization and project are required' }
|
|
}
|
|
|
|
const maxItems = sourceConfig.maxItems as string | undefined
|
|
if (maxItems && (Number.isNaN(Number(maxItems)) || Number(maxItems) <= 0)) {
|
|
return { valid: false, error: 'Max items must be a positive number' }
|
|
}
|
|
|
|
const customWiql = readString(sourceConfig.customWiql)
|
|
if (customWiql && !/from\s+workitems/i.test(customWiql)) {
|
|
return {
|
|
valid: false,
|
|
error: 'Custom WIQL query must select work items (e.g. "... FROM workitems WHERE ...")',
|
|
}
|
|
}
|
|
|
|
const contentType = parseContentType(sourceConfig.contentType)
|
|
const repositoryFilter = readString(sourceConfig.repositoryName)
|
|
|
|
try {
|
|
const url = `${ADO_BASE_URL}/${encodeURIComponent(organization)}/_apis/projects/${encodeURIComponent(project)}?api-version=${PROJECT_API_VERSION}`
|
|
const response = await fetchWithRetry(
|
|
url,
|
|
{
|
|
method: 'GET',
|
|
headers: { Accept: 'application/json', Authorization: patAuthHeader(accessToken) },
|
|
},
|
|
VALIDATE_RETRY_OPTIONS
|
|
)
|
|
|
|
if (response.status === 401 || response.status === 403) {
|
|
return { valid: false, error: 'Invalid or unauthorized Personal Access Token' }
|
|
}
|
|
if (response.status === 404) {
|
|
return {
|
|
valid: false,
|
|
error: `Project "${project}" not found in organization "${organization}"`,
|
|
}
|
|
}
|
|
if (!response.ok) {
|
|
return { valid: false, error: `Cannot access project: ${response.status}` }
|
|
}
|
|
|
|
if (activePhases(contentType).includes('file')) {
|
|
const repos = await listRepositories(
|
|
accessToken,
|
|
organization,
|
|
project,
|
|
VALIDATE_RETRY_OPTIONS
|
|
)
|
|
if (repositoryFilter) {
|
|
const needle = repositoryFilter.toLowerCase()
|
|
const match = repos.find(
|
|
(r) => r.id.toLowerCase() === needle || (r.name ?? '').toLowerCase() === needle
|
|
)
|
|
if (!match) {
|
|
return {
|
|
valid: false,
|
|
error: `Repository "${repositoryFilter}" not found in project "${project}"`,
|
|
}
|
|
}
|
|
if (match.isDisabled) {
|
|
return {
|
|
valid: false,
|
|
error: `Repository "${repositoryFilter}" is disabled`,
|
|
}
|
|
}
|
|
} else if (repos.length === 0) {
|
|
if (contentType === 'files') {
|
|
return {
|
|
valid: false,
|
|
error: `No accessible Git repositories found in project "${project}"`,
|
|
}
|
|
}
|
|
logger.warn('No accessible repositories; repository files will be skipped', {
|
|
organization,
|
|
project,
|
|
})
|
|
}
|
|
}
|
|
|
|
return { valid: true }
|
|
} catch (error) {
|
|
return { valid: false, error: getErrorMessage(error, 'Failed to validate configuration') }
|
|
}
|
|
},
|
|
|
|
/**
|
|
* Maps document metadata to tag slots. `kind` applies to every document.
|
|
* `wikiName` is wiki-only; `workItemType`/`state`/`areaPath`/`tags`/`changedDate`
|
|
* are work-item-only; `repository`/`path` are file-only. Each document type leaves
|
|
* the others' fields empty and the type/empty guards below skip them.
|
|
*/
|
|
mapTags: (metadata: Record<string, unknown>): Record<string, unknown> => {
|
|
const result: Record<string, unknown> = {}
|
|
|
|
if (typeof metadata.kind === 'string') result.kind = metadata.kind
|
|
if (typeof metadata.wikiName === 'string' && metadata.wikiName) {
|
|
result.wikiName = metadata.wikiName
|
|
}
|
|
if (typeof metadata.workItemType === 'string' && metadata.workItemType) {
|
|
result.workItemType = metadata.workItemType
|
|
}
|
|
if (typeof metadata.state === 'string' && metadata.state) result.state = metadata.state
|
|
if (typeof metadata.areaPath === 'string' && metadata.areaPath)
|
|
result.areaPath = metadata.areaPath
|
|
|
|
if (typeof metadata.repository === 'string' && metadata.repository) {
|
|
result.repository = metadata.repository
|
|
}
|
|
if (typeof metadata.path === 'string' && metadata.path) result.path = metadata.path
|
|
|
|
const tags = joinTagArray(metadata.tags)
|
|
if (tags) result.tags = tags
|
|
|
|
const changedDate = parseTagDate(metadata.changedDate)
|
|
if (changedDate) result.changedDate = changedDate
|
|
|
|
return result
|
|
},
|
|
}
|