Community connector PR board sync #219
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # Community PRs touching a certified API source land in Status "Community PR | |
| # Review"; community PRs that add a connector directory not present on master | |
| # land in Status "New Connector PRs", which takes precedence when both apply. | |
| # | |
| # Live synchronization is the default; scheduled runs always write. A | |
| # workflow_dispatch run can opt into log-only mode with dry_run=true, and the | |
| # max_writes input can cap a live run (0 means unlimited). Disable the workflow | |
| # from the Actions UI to stop synchronization entirely. The Octavia Bot app must | |
| # have org-level Projects: write permission for live synchronization. | |
| name: Community connector PR board sync | |
| on: | |
| schedule: | |
| - cron: "0 * * * *" | |
| workflow_dispatch: | |
| inputs: | |
| dry_run: | |
| description: "Log-only mode: read and report changes without writing to the project" | |
| required: false | |
| default: false | |
| type: boolean | |
| max_writes: | |
| description: "Maximum number of PRs to write to in this run; 0 means unlimited" | |
| required: false | |
| default: "0" | |
| type: string | |
| permissions: {} | |
| concurrency: | |
| group: ${{ github.workflow }} | |
| cancel-in-progress: false | |
| jobs: | |
| reconcile: | |
| name: Reconcile community connector PRs | |
| runs-on: ubuntu-24.04 | |
| timeout-minutes: 30 | |
| steps: | |
| - name: Resolve effective mode | |
| id: mode | |
| env: | |
| EVENT_NAME: ${{ github.event_name }} | |
| INPUT_DRY_RUN: ${{ inputs.dry_run }} | |
| MAX_WRITES_INPUT: ${{ inputs.max_writes }} | |
| run: | | |
| set -euo pipefail | |
| max_writes="${MAX_WRITES_INPUT:-0}" | |
| if [[ ! "$max_writes" =~ ^[0-9]+$ ]]; then | |
| echo "::error::max_writes must be a non-negative integer; received '$max_writes'." | |
| exit 1 | |
| fi | |
| if [[ "$EVENT_NAME" == "workflow_dispatch" && "$INPUT_DRY_RUN" == "true" ]]; then | |
| dry_run=true | |
| else | |
| dry_run=false | |
| fi | |
| echo "dry_run=$dry_run" >> "$GITHUB_OUTPUT" | |
| echo "::notice::Community connector PR board sync mode: $( [[ "$dry_run" == "true" ]] && echo DRY-RUN || echo LIVE WRITE )" | |
| # ---------- Authentication ---------- | |
| - name: Authenticate as Octavia Bot | |
| id: app-token | |
| uses: actions/create-github-app-token@f8d387b68d61c58ab83c6c016672934102569859 # v3.0.0 | |
| with: | |
| owner: airbytehq | |
| repositories: "airbyte" | |
| app-id: ${{ secrets.OCTAVIA_BOT_APP_ID }} | |
| private-key: ${{ secrets.OCTAVIA_BOT_PRIVATE_KEY }} | |
| - name: Checkout master | |
| uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2 | |
| with: | |
| ref: master # PR head code is never checked out or executed in a job holding the app token. | |
| token: ${{ steps.app-token.outputs.token }} # permissions: {} leaves GITHUB_TOKEN without contents: read. | |
| - name: Install yq | |
| run: sudo snap install yq | |
| - name: Build certified API source set | |
| id: certified-sources | |
| shell: bash | |
| run: | | |
| set -euo pipefail | |
| output="$RUNNER_TEMP/certified-api-sources.json" | |
| mapfile -d '' metadata_files < <( | |
| find airbyte-integrations/connectors -mindepth 2 -maxdepth 2 -name metadata.yaml -print0 | sort -z | |
| ) | |
| yq eval-all -o=json ' | |
| [ | |
| select( | |
| (.data.connectorType // "unknown") == "source" and | |
| (.data.connectorSubtype // "unknown") == "api" and | |
| (.data.supportLevel // "unknown") == "certified" | |
| ) | | |
| filename | split("/") | .[-2] | |
| ] | sort | unique | |
| ' "${metadata_files[@]}" > "$output" | |
| count="$(jq 'length' "$output")" | |
| echo "Certified API source connector count: $count" | |
| jq -r '.[]' "$output" | |
| if [[ "$count" -eq 0 ]]; then | |
| echo "::error::No certified API source connectors were found; refusing a silent no-op." | |
| exit 1 | |
| fi | |
| echo "path=$output" >> "$GITHUB_OUTPUT" | |
| - name: Build existing connector set | |
| id: existing-connectors | |
| shell: bash | |
| run: | | |
| set -euo pipefail | |
| output="$RUNNER_TEMP/existing-connectors.json" | |
| find airbyte-integrations/connectors -mindepth 1 -maxdepth 1 -type d -printf '%f\n' \ | |
| | sort -u \ | |
| | jq -R . | jq -sc . > "$output" | |
| count="$(jq 'length' "$output")" | |
| echo "Existing connector directory count: $count" | |
| if [[ "$count" -eq 0 ]]; then | |
| echo "::error::No connector directories were found; refusing a silent no-op." | |
| exit 1 | |
| fi | |
| echo "path=$output" >> "$GITHUB_OUTPUT" | |
| - name: Reconcile project board | |
| uses: actions/github-script@ed597411d8f924073f98dfc5c65a23a2325f34cd # v8.0.0 | |
| env: | |
| CERTIFIED_SOURCES_PATH: ${{ steps.certified-sources.outputs.path }} | |
| EXISTING_CONNECTORS_PATH: ${{ steps.existing-connectors.outputs.path }} | |
| DRY_RUN: ${{ steps.mode.outputs.dry_run }} | |
| MAX_WRITES: ${{ inputs.max_writes || '0' }} | |
| with: | |
| github-token: ${{ steps.app-token.outputs.token }} | |
| retries: 3 | |
| script: | | |
| const fs = require('fs'); | |
| const dryRun = process.env.DRY_RUN !== 'false'; | |
| const maxWritesInput = process.env.MAX_WRITES || '0'; | |
| if (!/^[0-9]+$/.test(maxWritesInput)) { | |
| throw new Error(`MAX_WRITES must be a non-negative integer; received "${maxWritesInput}".`); | |
| } | |
| const maxWrites = Number(maxWritesInput); | |
| if (!Number.isSafeInteger(maxWrites)) { | |
| throw new Error(`MAX_WRITES is too large to represent safely: "${maxWritesInput}".`); | |
| } | |
| const certifiedSources = new Set( | |
| JSON.parse(fs.readFileSync(process.env.CERTIFIED_SOURCES_PATH, 'utf8')), | |
| ); | |
| const existingConnectors = new Set( | |
| JSON.parse(fs.readFileSync(process.env.EXISTING_CONNECTORS_PATH, 'utf8')), | |
| ); | |
| const counts = { | |
| evaluated: 0, | |
| qualified: 0, | |
| added: 0, | |
| statusSet: 0, | |
| draft: 0, | |
| notCommunity: 0, | |
| notQualifying: 0, | |
| statusAlreadySet: 0, | |
| writeCapReached: 0, | |
| certifiedApiSource: 0, | |
| newConnector: 0, | |
| }; | |
| const failures = []; | |
| const projectNumber = 137; | |
| const repositoryVariables = { owner: 'airbytehq', name: 'airbyte' }; | |
| let writesPerformed = 0; | |
| const projectQuery = ` | |
| query($login: String!, $number: Int!) { | |
| organization(login: $login) { | |
| projectV2(number: $number) { | |
| id | |
| title | |
| field(name: "Status") { | |
| ... on ProjectV2SingleSelectField { | |
| id | |
| options { id name } | |
| } | |
| } | |
| } | |
| } | |
| } | |
| `; | |
| let project; | |
| try { | |
| const result = await github.graphql(projectQuery, { | |
| login: 'airbytehq', | |
| number: projectNumber, | |
| }); | |
| project = result.organization?.projectV2; | |
| } catch (error) { | |
| throw new Error( | |
| `Could not resolve organization project ${projectNumber}. ` + | |
| 'The app token may lack org-level Projects read/write permission. ' + | |
| `GraphQL error: ${error.message}`, | |
| ); | |
| } | |
| if (!project) { | |
| throw new Error( | |
| `Organization project ${projectNumber} was not found. ` + | |
| 'Check that it exists and that the app token has org-level Projects read/write permission.', | |
| ); | |
| } | |
| if (!project.field?.id || !Array.isArray(project.field.options)) { | |
| throw new Error( | |
| `Project "${project.title}" has no Status single-select field. ` + | |
| 'Check the project configuration and that the app token has org-level Projects read/write permission.', | |
| ); | |
| } | |
| const findStatusOption = (name) => { | |
| const option = project.field.options.find((candidate) => candidate.name === name); | |
| if (!option) { | |
| throw new Error( | |
| `Project "${project.title}" has no Status option named exactly "${name}". ` + | |
| 'Check the project configuration and that the app token has org-level Projects read/write permission.', | |
| ); | |
| } | |
| return option; | |
| }; | |
| const certifiedStatusOption = findStatusOption('Community PR Review'); | |
| const newConnectorStatusOption = findStatusOption('New Connector PRs'); | |
| core.info( | |
| `Resolved project "${project.title}" with options ` + | |
| `"${certifiedStatusOption.name}" and "${newConnectorStatusOption.name}"`, | |
| ); | |
| const pullRequestsQuery = ` | |
| query($owner: String!, $name: String!, $cursor: String) { | |
| repository(owner: $owner, name: $name) { | |
| pullRequests( | |
| states: OPEN | |
| first: 50 | |
| after: $cursor | |
| orderBy: { field: CREATED_AT, direction: ASC } | |
| ) { | |
| pageInfo { hasNextPage endCursor } | |
| nodes { | |
| id number isDraft authorAssociation | |
| headRepositoryOwner { login } | |
| changedFiles | |
| files(first: 100) { | |
| pageInfo { hasNextPage endCursor } | |
| nodes { path } | |
| } | |
| } | |
| } | |
| } | |
| } | |
| `; | |
| const additionalFilesQuery = ` | |
| query($owner: String!, $name: String!, $number: Int!, $cursor: String) { | |
| repository(owner: $owner, name: $name) { | |
| pullRequest(number: $number) { | |
| files(first: 100, after: $cursor) { | |
| pageInfo { hasNextPage endCursor } | |
| nodes { path } | |
| } | |
| } | |
| } | |
| } | |
| `; | |
| const projectItemsQuery = ` | |
| query($id: ID!, $cursor: String) { | |
| node(id: $id) { | |
| ... on PullRequest { | |
| projectItems(first: 50, after: $cursor) { | |
| pageInfo { hasNextPage endCursor } | |
| nodes { | |
| id | |
| project { number } | |
| fieldValueByName(name: "Status") { | |
| ... on ProjectV2ItemFieldSingleSelectValue { name } | |
| } | |
| } | |
| } | |
| } | |
| } | |
| } | |
| `; | |
| const addItemMutation = ` | |
| mutation($projectId: ID!, $contentId: ID!) { | |
| addProjectV2ItemById(input: { projectId: $projectId, contentId: $contentId }) { | |
| item { id } | |
| } | |
| } | |
| `; | |
| const updateStatusMutation = ` | |
| mutation($projectId: ID!, $itemId: ID!, $fieldId: ID!, $optionId: String!) { | |
| updateProjectV2ItemFieldValue(input: { | |
| projectId: $projectId | |
| itemId: $itemId | |
| fieldId: $fieldId | |
| value: { singleSelectOptionId: $optionId } | |
| }) { | |
| projectV2Item { id } | |
| } | |
| } | |
| `; | |
| const connectorNameFromPath = (path) => { | |
| const match = (path ?? '').match(/^airbyte-integrations\/connectors\/([^/]+)\//); | |
| return match ? match[1] : null; | |
| }; | |
| const matchesCertifiedSource = (path) => { | |
| const connector = connectorNameFromPath(path); | |
| return Boolean(connector) && certifiedSources.has(connector); | |
| }; | |
| const matchesNewConnector = (path) => { | |
| const connector = connectorNameFromPath(path); | |
| return Boolean(connector) && !existingConnectors.has(connector); | |
| }; | |
| const classify = (files) => { | |
| if (files.some((file) => matchesNewConnector(file.path))) { | |
| return { kind: 'new-connector', statusOption: newConnectorStatusOption }; | |
| } | |
| if (files.some((file) => matchesCertifiedSource(file.path))) { | |
| return { kind: 'certified-api-source', statusOption: certifiedStatusOption }; | |
| } | |
| return null; | |
| }; | |
| const getAllFiles = async (pr) => { | |
| const files = [...(pr.files?.nodes ?? [])]; | |
| const initialPageInfo = pr.files?.pageInfo ?? {}; | |
| let cursor = initialPageInfo.endCursor ?? null; | |
| let hasNextPage = pr.changedFiles > 100 || Boolean(initialPageInfo.hasNextPage); | |
| while (!files.some((file) => matchesNewConnector(file.path)) && hasNextPage) { | |
| const result = await github.graphql(additionalFilesQuery, { | |
| ...repositoryVariables, | |
| number: pr.number, | |
| cursor, | |
| }); | |
| const page = result.repository.pullRequest?.files; | |
| if (!page) break; | |
| files.push(...(page.nodes ?? [])); | |
| cursor = page.pageInfo?.endCursor ?? null; | |
| hasNextPage = Boolean(page.pageInfo?.hasNextPage); | |
| } | |
| return files; | |
| }; | |
| const getProjectItem = async (pr) => { | |
| const items = []; | |
| let cursor = null; | |
| let hasNextPage = true; | |
| while (hasNextPage) { | |
| const result = await github.graphql(projectItemsQuery, { id: pr.id, cursor }); | |
| const page = result.node.projectItems; | |
| items.push(...page.nodes); | |
| hasNextPage = page.pageInfo.hasNextPage; | |
| cursor = page.pageInfo.endCursor; | |
| } | |
| return items.find((item) => item.project?.number === projectNumber); | |
| }; | |
| const writeCapReached = () => !dryRun && maxWrites > 0 && writesPerformed >= maxWrites; | |
| const writeCapDescription = maxWrites > 0 ? ` (max_writes=${maxWrites})` : ''; | |
| if (dryRun && maxWrites > 0) { | |
| core.info(`Dry-run: max_writes=${maxWrites} is configured but ignored.`); | |
| } | |
| let projectTitle = project.title; | |
| const writeSummary = async () => { | |
| const mode = dryRun ? 'DRY-RUN' : 'LIVE'; | |
| await core.summary | |
| .addHeading('Community connector PR board sync') | |
| .addTable([ | |
| [{ data: 'Metric', header: true }, { data: 'Count', header: true }], | |
| ['Mode', mode], | |
| ['Project', projectTitle], | |
| ['Status option (certified API source)', certifiedStatusOption.name], | |
| ['Status option (new connector)', newConnectorStatusOption.name], | |
| ['Max writes', dryRun ? `Ignored${writeCapDescription}` : (maxWrites > 0 ? String(maxWrites) : 'Unlimited')], | |
| ['Evaluated', String(counts.evaluated)], | |
| ['Qualified', String(counts.qualified)], | |
| ['Qualified: certified API source change', String(counts.certifiedApiSource)], | |
| ['Qualified: new connector', String(counts.newConnector)], | |
| [dryRun ? 'Would add' : 'Added', String(counts.added)], | |
| [dryRun ? 'Would set Status' : 'Status-set', String(counts.statusSet)], | |
| ['Skipped: draft', String(counts.draft)], | |
| ['Skipped: not-community', String(counts.notCommunity)], | |
| ['Skipped: no-qualifying-connector-change', String(counts.notQualifying)], | |
| ['Skipped: status-already-set', String(counts.statusAlreadySet)], | |
| ['Skipped: write cap reached', String(counts.writeCapReached)], | |
| ]) | |
| .addHeading('Mutation failures') | |
| .addList(failures.length ? failures : ['None']) | |
| .write(); | |
| }; | |
| let cursor = null; | |
| let hasNextPage = true; | |
| try { | |
| while (hasNextPage) { | |
| const result = await github.graphql(pullRequestsQuery, { | |
| ...repositoryVariables, | |
| cursor, | |
| }); | |
| const page = result.repository.pullRequests; | |
| for (const pr of page.nodes) { | |
| counts.evaluated += 1; | |
| if (pr.isDraft) { | |
| counts.draft += 1; | |
| core.info(`PR ${pr.number}: skipped (draft)`); | |
| continue; | |
| } | |
| const isCommunity = | |
| pr.headRepositoryOwner?.login !== 'airbytehq' && | |
| !['MEMBER', 'OWNER', 'COLLABORATOR'].includes(pr.authorAssociation); | |
| if (!isCommunity) { | |
| counts.notCommunity += 1; | |
| core.info(`PR ${pr.number}: skipped (not-community)`); | |
| continue; | |
| } | |
| try { | |
| const files = await getAllFiles(pr); | |
| const classification = classify(files); | |
| if (!classification) { | |
| counts.notQualifying += 1; | |
| core.info(`PR ${pr.number}: skipped (no-qualifying-connector-change)`); | |
| continue; | |
| } | |
| const statusOption = classification.statusOption; | |
| counts.qualified += 1; | |
| counts[classification.kind === 'new-connector' ? 'newConnector' : 'certifiedApiSource'] += 1; | |
| core.info(`PR ${pr.number}: qualified (${classification.kind})`); | |
| if (writeCapReached()) { | |
| counts.writeCapReached += 1; | |
| core.info(`PR ${pr.number}: skipped (write cap reached${writeCapDescription})`); | |
| continue; | |
| } | |
| const item = await getProjectItem(pr); | |
| if (!item) { | |
| if (dryRun) { | |
| counts.added += 1; | |
| counts.statusSet += 1; | |
| core.info(`PR ${pr.number}: would add to project and set Status="${statusOption.name}"`); | |
| } else { | |
| const added = await github.graphql(addItemMutation, { | |
| projectId: project.id, | |
| contentId: pr.id, | |
| }); | |
| counts.added += 1; | |
| writesPerformed += 1; | |
| await github.graphql(updateStatusMutation, { | |
| projectId: project.id, | |
| itemId: added.addProjectV2ItemById.item.id, | |
| fieldId: project.field.id, | |
| optionId: statusOption.id, | |
| }); | |
| counts.statusSet += 1; | |
| core.info(`PR ${pr.number}: added to project and set Status="${statusOption.name}"`); | |
| } | |
| } else if (!item.fieldValueByName?.name) { | |
| if (dryRun) { | |
| counts.statusSet += 1; | |
| core.info(`PR ${pr.number}: would set empty Status to "${statusOption.name}"`); | |
| } else { | |
| await github.graphql(updateStatusMutation, { | |
| projectId: project.id, | |
| itemId: item.id, | |
| fieldId: project.field.id, | |
| optionId: statusOption.id, | |
| }); | |
| counts.statusSet += 1; | |
| writesPerformed += 1; | |
| core.info(`PR ${pr.number}: set empty Status to "${statusOption.name}"`); | |
| } | |
| } else { | |
| counts.statusAlreadySet += 1; | |
| core.info(`PR ${pr.number}: skipped (status-already-set: "${item.fieldValueByName.name}")`); | |
| } | |
| } catch (error) { | |
| failures.push(`PR ${pr.number}: ${error.message}`); | |
| core.error(`PR ${pr.number}: reconciliation failed: ${error.message}`); | |
| } | |
| } | |
| hasNextPage = page.pageInfo.hasNextPage; | |
| cursor = page.pageInfo.endCursor; | |
| } | |
| } catch (error) { | |
| failures.push(`Open PR page fetch: ${error.message}`); | |
| core.error(`Open PR page fetch failed: ${error.message}`); | |
| await writeSummary(); | |
| core.setFailed(`Open PR scan failed after collecting available results: ${error.message}`); | |
| return; | |
| } | |
| await writeSummary(); | |
| if (failures.length) { | |
| core.setFailed(`${failures.length} PR reconciliation(s) failed.`); | |
| } |