diff --git a/backend/src/osspckgs/migrations/V1781539311__packages_tables_sequin_updates.sql b/backend/src/osspckgs/migrations/V1781539311__packages_tables_sequin_updates.sql new file mode 100644 index 0000000000..d84c330567 --- /dev/null +++ b/backend/src/osspckgs/migrations/V1781539311__packages_tables_sequin_updates.sql @@ -0,0 +1,30 @@ +-- ─── 1. package_dependencies (created_at, id) index ───────────────────────── +CREATE INDEX IF NOT EXISTS package_dependencies_created_at_id_idx + ON package_dependencies (created_at, id); + +-- ─── 2. repo_activity_snapshot replication ────────────────────────────────── +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM pg_publication WHERE pubname = 'sequin_pub' + ) AND NOT EXISTS ( + SELECT 1 FROM pg_publication_tables + WHERE pubname = 'sequin_pub' + AND schemaname = 'public' + AND tablename = 'repo_activity_snapshot' + ) THEN + ALTER PUBLICATION sequin_pub ADD TABLE repo_activity_snapshot; + END IF; +END$$; + +ALTER TABLE public.repo_activity_snapshot REPLICA IDENTITY FULL; + +-- ─── 3. packages enriched health/lifecycle fields (synced back from Tinybird) ─── +ALTER TABLE packages + ADD COLUMN IF NOT EXISTS lifecycle_label text, + ADD COLUMN IF NOT EXISTS health_score smallint, + ADD COLUMN IF NOT EXISTS health_label text, + ADD COLUMN IF NOT EXISTS maintainer_health_score smallint, + ADD COLUMN IF NOT EXISTS security_supply_chain_score smallint, + ADD COLUMN IF NOT EXISTS development_activity_score smallint, + ADD COLUMN IF NOT EXISTS signal_coverage_health jsonb; diff --git a/services/libs/tinybird/datasources/ossPackages_enriched_ds.datasource b/services/libs/tinybird/datasources/ossPackages_enriched_ds.datasource new file mode 100644 index 0000000000..da1ac01c7b --- /dev/null +++ b/services/libs/tinybird/datasources/ossPackages_enriched_ds.datasource @@ -0,0 +1,58 @@ +DESCRIPTION > + - `ossPackages_enriched_ds` contains all `ossPackages` metadata plus derived enrichment fields, materialized by `ossPackages_enriched.pipe` (daily COPY, replace mode). + - Carries every column from the `ossPackages` datasource verbatim, so it can serve as a drop-in enriched replacement for package analytics. + - `lifecycleLabel` is the categorical maintenance state of the package's primary repository, derived from repo + activity-snapshot signals: 'active', 'stable', 'declining', 'abandoned', or 'archived'. + - `healthScore` is the composite 0–100 health score = `maintainerHealthScore` (≤40) + `securitySupplyChainScore` (≤35) + `developmentActivityScore` (≤25). + - `healthLabel` buckets `healthScore`: 'excellent' (85+), 'healthy' (70–84), 'fair' (50–69), 'concerning' (30–49), 'critical' (<30). + - `maintainerHealthScore`, `securitySupplyChainScore`, `developmentActivityScore` are the three health sub-totals. + - `signalCoverageHealth` is a JSON object mapping each health sub-signal to 'available' | 'partial' | 'blocked', so the UI can flag which scores reflect real data vs. gaps. + - `impact` (carried verbatim from `ossPackages`) is the criticality impact score and serves as the Impact total score. + - All other fields mirror `ossPackages`; see `ossPackages.datasource` for per-field documentation. + +TAGS "OSS packages", "Enrichment" + +SCHEMA > + `id` UInt64, + `purl` String, + `ecosystem` String, + `namespace` Nullable(String), + `name` String, + `registryUrl` Nullable(String), + `status` Nullable(String), + `description` Nullable(String), + `homepage` Nullable(String), + `declaredRepositoryUrl` Nullable(String), + `repositoryUrl` Nullable(String), + `licenses` Array(String), + `licensesRaw` Nullable(String), + `keywords` Array(String), + `distTagsLatest` Nullable(String), + `distTagsNext` Nullable(String), + `distTagsBeta` Nullable(String), + `versionsCount` Nullable(UInt32), + `latestVersion` Nullable(String), + `firstReleaseAt` Nullable(DateTime64(3)), + `latestReleaseAt` Nullable(DateTime64(3)), + `dependentCount` Nullable(UInt32), + `transitiveDependentCount` Nullable(UInt64), + `dependentReposCount` Nullable(UInt32), + `hasCriticalVulnerability` UInt8, + `impact` Nullable(String), + `isCritical` UInt8, + `lastRankPassAt` Nullable(DateTime64(3)), + `downloadsLast30d` Nullable(UInt64), + `centralityScore` Nullable(String), + `rankInEcosystem` Nullable(UInt32), + `ingestionSource` Nullable(String), + `lastSyncedAt` DateTime64(3), + `createdAt` DateTime64(3), + `lifecycleLabel` String, + `healthScore` UInt8, + `healthLabel` String, + `maintainerHealthScore` UInt8, + `securitySupplyChainScore` UInt8, + `developmentActivityScore` UInt8, + `signalCoverageHealth` String + +ENGINE MergeTree +ENGINE_SORTING_KEY ecosystem, purl diff --git a/services/libs/tinybird/datasources/repoActivitySnapshot.datasource b/services/libs/tinybird/datasources/repoActivitySnapshot.datasource new file mode 100644 index 0000000000..6375f1d2c1 --- /dev/null +++ b/services/libs/tinybird/datasources/repoActivitySnapshot.datasource @@ -0,0 +1,41 @@ +DESCRIPTION > + - `repoActivitySnapshot` holds per-repository activity health metrics over a rolling window (default 12 months). + - Replicated from Postgres packages-db — computed by the github-repos-enricher worker from the GitHub API (commits, PRs, issues). + - One row per repo; a row is absent until the enricher has reached that repo, so consumers must treat a missing row as "no signal" (degraded path). + - Used to derive package lifecycle (active / stable / declining / abandoned) by joining repos -> packageRepos -> packages. + - `repoId` links to the parent repos row — the primary key. + - `snapshotAt` is when this snapshot was computed — also the updatedAt watermark for Tinybird sync. + - `windowMonths` is the rolling window length the metrics were computed over (default 12). + - `commitsLast12m`, `commitsLast6m`, `commitsPrior6m` are commit counts for the trailing 12m, the most recent 6m, and the 6m before that (NULL if not computed). + - `prsOpenedLast12m`, `prsMergedLast12m`, `prsClosedUnmerged12m` are PR counts opened / merged / closed-without-merge in the trailing 12m (NULL if not computed). + - `prMedianTimeToMergeHours` is the median hours from PR open to merge (NULL if no merged PRs in window). + - `prMedianTimeToFirstResponseHours` is the median hours to first non-author response on a PR (NULL if no PRs or no responses in window). + - `issuesOpenedLast12m`, `issuesClosedLast12m` are issue counts opened / closed in the trailing 12m (NULL if not computed). + - `issuesOpenedLast6m`, `issuesOpenedPrior6m` are issues opened in the most recent 6m and the 6m before that (NULL if not computed). + - `issuesOpenNow` is the current count of open issues (NULL if not computed). + - `issueMedianTimeToCloseHours` is the median hours from issue open to close (NULL if no closed issues in window). + - `issueMedianTimeToFirstResponseHours` is the median hours to first non-author response on an issue (NULL if no issues or no responses in window). + +SCHEMA > + `repoId` UInt64 `json:$.record.repo_id`, + `snapshotAt` DateTime64(3) `json:$.record.snapshot_at`, + `windowMonths` UInt16 `json:$.record.window_months` DEFAULT 12, + `commitsLast12m` Nullable(Int32) `json:$.record.commits_last_12m`, + `commitsLast6m` Nullable(Int32) `json:$.record.commits_last_6m`, + `commitsPrior6m` Nullable(Int32) `json:$.record.commits_prior_6m`, + `prsOpenedLast12m` Nullable(Int32) `json:$.record.prs_opened_last_12m`, + `prsMergedLast12m` Nullable(Int32) `json:$.record.prs_merged_last_12m`, + `prsClosedUnmerged12m` Nullable(Int32) `json:$.record.prs_closed_unmerged_12m`, + `prMedianTimeToMergeHours` Nullable(Int32) `json:$.record.pr_median_time_to_merge_hours`, + `prMedianTimeToFirstResponseHours` Nullable(Int32) `json:$.record.pr_median_time_to_first_response_hours`, + `issuesOpenedLast12m` Nullable(Int32) `json:$.record.issues_opened_last_12m`, + `issuesClosedLast12m` Nullable(Int32) `json:$.record.issues_closed_last_12m`, + `issuesOpenedLast6m` Nullable(Int32) `json:$.record.issues_opened_last_6m`, + `issuesOpenedPrior6m` Nullable(Int32) `json:$.record.issues_opened_prior_6m`, + `issuesOpenNow` Nullable(Int32) `json:$.record.issues_open_now`, + `issueMedianTimeToCloseHours` Nullable(Int32) `json:$.record.issue_median_time_to_close_hours`, + `issueMedianTimeToFirstResponseHours` Nullable(Int32) `json:$.record.issue_median_time_to_first_response_hours` + +ENGINE ReplacingMergeTree +ENGINE_SORTING_KEY repoId +ENGINE_VER snapshotAt diff --git a/services/libs/tinybird/pipes/ossPackages_enriched.pipe b/services/libs/tinybird/pipes/ossPackages_enriched.pipe new file mode 100644 index 0000000000..6140ae63d0 --- /dev/null +++ b/services/libs/tinybird/pipes/ossPackages_enriched.pipe @@ -0,0 +1,440 @@ +NODE ossPackages_enriched_packages +SQL > + SELECT * FROM ossPackages FINAL + +NODE ossPackages_enriched_repos +SQL > + SELECT + id AS repoId, + archived, + lastCommitAt, + scorecardScore, + securityPolicyEnabled, + branchProtectionEnabled, + branchProtectionRequiredReviews, + branchProtectionRequiresStatusChecks, + branchProtectionAllowsForcePush + FROM repos FINAL + +NODE ossPackages_enriched_pkg_repo +SQL > + SELECT packageId, argMax(repoId, (toFloat64OrZero(confidence), verifiedAt)) AS repoId, 1 AS hasRepo + FROM packageRepos FINAL + GROUP BY packageId + +NODE ossPackages_enriched_snapshot +SQL > + SELECT + repoId, + 1 AS hasSnapshot, + commitsLast6m, + commitsPrior6m, + issuesOpenedLast6m, + issuesOpenedPrior6m, + issuesOpenedLast12m, + issuesClosedLast12m, + prsOpenedLast12m, + prsMergedLast12m, + prsClosedUnmerged12m, + issuesOpenNow, + issueMedianTimeToCloseHours, + prMedianTimeToMergeHours, + prMedianTimeToFirstResponseHours, + multiIf( + prMedianTimeToFirstResponseHours IS NOT NULL + AND issueMedianTimeToFirstResponseHours IS NOT NULL, + (prMedianTimeToFirstResponseHours + issueMedianTimeToFirstResponseHours) / 2, + prMedianTimeToFirstResponseHours IS NOT NULL, + prMedianTimeToFirstResponseHours, + issueMedianTimeToFirstResponseHours IS NOT NULL, + issueMedianTimeToFirstResponseHours, + NULL + ) AS combinedResponseHours + FROM repoActivitySnapshot FINAL + +NODE ossPackages_enriched_maintainers +SQL > + SELECT + pm.packageId AS packageId, + count() AS maintainersCount, + uniqExactIf( + lower(arrayElement(splitByChar('@', m.email), -1)), + m.email != '' AND position(m.email, '@') > 0 + ) AS emailDomains + FROM packageMaintainers AS pm FINAL + LEFT JOIN maintainers AS m FINAL ON pm.maintainerId = m.id + GROUP BY pm.packageId + +NODE ossPackages_enriched_release +SQL > + SELECT + packageId, + max(publishedAt) AS maxPublished, + arrayElement( + arraySort(x -> (- toUnixTimestamp(x)), groupArray(publishedAt)), 2 + ) AS secondPublished + FROM versions FINAL + WHERE isNotNull(publishedAt) + GROUP BY packageId + +NODE ossPackages_enriched_open_vulns +SQL > + SELECT + ap.packageId AS packageId, + uniqExactIf(a.id, lower(a.severity) = 'critical') AS criticalOpen, + uniqExactIf(a.id, lower(a.severity) = 'high') AS highOpen, + uniqExactIf(a.id, lower(a.severity) = 'medium') AS mediumOpen, + max( + (aar.introducedVersion != '' AND vi.introducedAt IS NULL) + OR (aar.fixedVersion != '' AND vf.fixedAt IS NULL) + OR (aar.lastAffected != '' AND vl.lastAffectedAt IS NULL) + ) AS boundUnresolved + FROM ossPackages_enriched_packages AS p + INNER JOIN advisoryPackages AS ap FINAL ON ap.packageId = p.id + INNER JOIN advisories AS a FINAL ON a.id = ap.advisoryId + INNER JOIN advisoryAffectedRanges AS aar FINAL ON aar.advisoryPackageId = ap.id + LEFT JOIN + (SELECT packageId, number, publishedAt AS introducedAt FROM versions FINAL) AS vi + ON vi.packageId = p.id + AND vi.number = aar.introducedVersion + LEFT JOIN + (SELECT packageId, number, publishedAt AS fixedAt FROM versions FINAL) AS vf + ON vf.packageId = p.id + AND vf.number = aar.fixedVersion + LEFT JOIN + (SELECT packageId, number, publishedAt AS lastAffectedAt FROM versions FINAL) AS vl + ON vl.packageId = p.id + AND vl.number = aar.lastAffected + WHERE + p.latestReleaseAt IS NOT NULL + AND ( + aar.introducedVersion = '' + OR vi.introducedAt IS NULL + OR p.latestReleaseAt >= vi.introducedAt + ) + AND (aar.fixedVersion = '' OR vf.fixedAt IS NULL OR p.latestReleaseAt < vf.fixedAt) + AND ( + aar.lastAffected = '' OR vl.lastAffectedAt IS NULL OR p.latestReleaseAt <= vl.lastAffectedAt + ) + GROUP BY ap.packageId + +NODE ossPackages_enriched_dep_vulns +SQL > + SELECT + pd.packageId AS packageId, + uniqExactIf(pd.dependsOnId, dep.hasCriticalVulnerability = 1) AS vulnerableDeps, + 1 AS hasDeps + FROM packageDependencies AS pd FINAL + INNER JOIN + (SELECT id, hasCriticalVulnerability FROM ossPackages_enriched_packages) AS dep + ON pd.dependsOnId = dep.id + WHERE pd.dependencyKind = 'direct' + GROUP BY pd.packageId + +NODE ossPackages_enriched_repo_orgs +SQL > + SELECT url, max(organizationCount) AS organizationCount, 1 AS inRepositories + FROM repositories_populated_ds + GROUP BY url + +NODE ossPackages_enriched_scored +SQL > + SELECT + p.*, + multiIf( + coalesce(r.archived, 0) = 1, + 'archived', + coalesce(snap.hasSnapshot, 0) = 0, + 'active', + ( + isNotNull(r.lastCommitAt) + AND r.lastCommitAt < subtractMonths(now(), 18) + AND ( + snap.prMedianTimeToFirstResponseHours IS NULL + OR ( + coalesce(snap.issuesOpenedLast12m, 0) = 0 + AND coalesce(snap.prsOpenedLast12m, 0) = 0 + ) + ) + ), + 'abandoned', + ( + coalesce(snap.commitsLast6m, 0) < coalesce(snap.commitsPrior6m, 0) * 0.5 + AND coalesce(snap.issuesOpenedLast6m, 0) > coalesce(snap.issuesOpenedPrior6m, 0) + ), + 'declining', + ( + isNotNull(p.latestReleaseAt) + AND p.latestReleaseAt > subtractMonths(now(), 12) + AND coalesce(snap.issuesOpenNow, 0) < 50 + AND p.hasCriticalVulnerability = 0 + AND coalesce(snap.commitsLast6m, 0) < coalesce(snap.commitsPrior6m, 0) * 0.5 + ), + 'stable', + 'active' + ) AS lifecycleLabel, + -- Maintainer Health (<=40) + multiIf( + coalesce(mh.maintainersCount, 0) >= 5, + 18, + coalesce(mh.maintainersCount, 0) >= 3, + 15, + coalesce(mh.maintainersCount, 0) = 2, + 6, + coalesce(mh.maintainersCount, 0) = 1, + 3, + 0 + ) AS busFactor, + multiIf( + coalesce(ro.inRepositories, 0) = 1, + multiIf( + coalesce(ro.organizationCount, 0) >= 3, + 7, + coalesce(ro.organizationCount, 0) = 2, + 5, + coalesce(ro.organizationCount, 0) = 1, + 2, + 0 + ), + multiIf( + coalesce(mh.emailDomains, 0) >= 3, + 7, + coalesce(mh.emailDomains, 0) = 2, + 5, + coalesce(mh.emailDomains, 0) = 1, + 2, + 0 + ) + ) AS orgDiversity, + multiIf( + snap.combinedResponseHours IS NULL, + 0, + snap.combinedResponseHours < 24, + 15, + snap.combinedResponseHours < 72, + 12, + snap.combinedResponseHours < 168, + 9, + snap.combinedResponseHours < 720, + 5, + snap.combinedResponseHours < 2160, + 2, + 0 + ) AS responsiveness, + toUInt8(busFactor + orgDiversity + responsiveness) AS maintainerHealthScore, + -- Security & Supply Chain (<=35) + if( + p.latestReleaseAt IS NULL, + 0, + greatest( + 0, + 10 + - 6 * coalesce(ov.criticalOpen, 0) + - 3 * coalesce(ov.highOpen, 0) + - coalesce(ov.mediumOpen, 0) + ) + ) AS openVulns, + round(toFloat64OrZero(r.scorecardScore) * 0.7) AS scorecardSub, + multiIf( + coalesce(dv.vulnerableDeps, 0) = 0, + 5, + coalesce(dv.vulnerableDeps, 0) <= 2, + 3, + coalesce(dv.vulnerableDeps, 0) <= 5, + 1, + 0 + ) AS dependencyHealth, + ( + (coalesce(r.securityPolicyEnabled, 0) = 1) * 2 + + (coalesce(r.branchProtectionEnabled, 0) = 1) * 2 + + (coalesce(r.branchProtectionRequiredReviews, 0) >= 1) * 1 + + (coalesce(r.branchProtectionRequiresStatusChecks, 0) = 1) * 1 + + (coalesce(r.branchProtectionAllowsForcePush, 1) = 0) * 1 + ) AS securityPractices, + 0 AS supplyChainIntegrity, + toUInt8( + openVulns + scorecardSub + dependencyHealth + securityPractices + supplyChainIntegrity + ) AS securitySupplyChainScore, + -- Development Activity (<=25) + multiIf( + rel.maxPublished IS NULL, + 0, + dateDiff('day', rel.maxPublished, now()) < 90 + AND dateDiff('day', rel.secondPublished, rel.maxPublished) < 90, + 8, + dateDiff('day', rel.maxPublished, now()) < 180, + 6, + dateDiff('day', rel.maxPublished, now()) < 365, + 4, + dateDiff('day', rel.maxPublished, now()) < 730, + 2, + 0 + ) AS releaseCadence, + multiIf( + coalesce(snap.hasSnapshot, 0) = 1, + multiIf( + coalesce(snap.commitsLast6m, 0) >= 50, + 5, + coalesce(snap.commitsLast6m, 0) >= 20, + 4, + coalesce(snap.commitsLast6m, 0) >= 5, + 3, + coalesce(snap.commitsLast6m, 0) >= 1, + 1, + 0 + ), + isNotNull(r.lastCommitAt) AND r.lastCommitAt > subtractMonths(now(), 3), + 4, + isNotNull(r.lastCommitAt) AND r.lastCommitAt > subtractMonths(now(), 12), + 2, + 0 + ) AS commitActivity, + ( + multiIf( + coalesce(snap.issuesOpenedLast12m, 0) = 0, + 0, + snap.issuesClosedLast12m / snap.issuesOpenedLast12m >= 0.8, + 4, + snap.issuesClosedLast12m / snap.issuesOpenedLast12m >= 0.5, + 2, + 0 + ) + multiIf( + snap.issueMedianTimeToCloseHours IS NULL, + 0, + snap.issueMedianTimeToCloseHours < 168, + 3, + snap.issueMedianTimeToCloseHours < 720, + 2, + 0 + ) + ) AS issueResolution, + ( + multiIf( + coalesce(snap.prsMergedLast12m, 0) + coalesce(snap.prsClosedUnmerged12m, 0) = 0, + 0, + snap.prsMergedLast12m / (snap.prsMergedLast12m + snap.prsClosedUnmerged12m) >= 0.7, + 3, + snap.prsMergedLast12m / (snap.prsMergedLast12m + snap.prsClosedUnmerged12m) >= 0.4, + 1, + 0 + ) + multiIf( + snap.prMedianTimeToMergeHours IS NULL, + 0, + snap.prMedianTimeToMergeHours < 168, + 2, + snap.prMedianTimeToMergeHours < 720, + 1, + 0 + ) + ) AS prMergeHealth, + toUInt8( + releaseCadence + commitActivity + issueResolution + prMergeHealth + ) AS developmentActivityScore, + toUInt8( + maintainerHealthScore + securitySupplyChainScore + developmentActivityScore + ) AS healthScore, + multiIf( + healthScore >= 85, + 'excellent', + healthScore >= 70, + 'healthy', + healthScore >= 50, + 'fair', + healthScore >= 30, + 'concerning', + 'critical' + ) AS healthLabel, + toJSONString( + map( + 'bus_factor', + if(coalesce(mh.maintainersCount, 0) > 0, 'available', 'blocked'), + 'org_diversity', + multiIf( + coalesce(ro.inRepositories, 0) = 1, + 'available', + coalesce(mh.emailDomains, 0) > 0, + 'partial', + 'blocked' + ), + 'responsiveness', + if(snap.combinedResponseHours IS NULL, 'blocked', 'available'), + 'open_vulnerabilities', + multiIf( + p.latestReleaseAt IS NULL, + 'blocked', + coalesce(ov.boundUnresolved, 0) = 1, + 'partial', + 'available' + ), + 'scorecard', + if(toFloat64OrZero(r.scorecardScore) > 0, 'available', 'blocked'), + 'dependency_health', + if(coalesce(dv.hasDeps, 0) = 1, 'available', 'blocked'), + 'security_practices', + multiIf( + r.repoId IS NULL, + 'blocked', + r.branchProtectionEnabled IS NULL, + 'partial', + 'available' + ), + 'supply_chain_integrity', + 'blocked', + 'release_cadence', + if(rel.maxPublished IS NULL, 'blocked', 'available'), + 'commit_activity', + multiIf( + coalesce(snap.hasSnapshot, 0) = 1, + 'available', + isNotNull(r.lastCommitAt), + 'partial', + 'blocked' + ), + 'issue_resolution', + if( + coalesce(snap.hasSnapshot, 0) = 1 AND coalesce(snap.issuesOpenedLast12m, 0) > 0, + 'available', + 'blocked' + ), + 'pr_merge_health', + if( + coalesce(snap.hasSnapshot, 0) = 1 + AND coalesce(snap.prsMergedLast12m, 0) + coalesce(snap.prsClosedUnmerged12m, 0) > 0, + 'available', + 'blocked' + ) + ) + ) AS signalCoverageHealth + FROM ossPackages_enriched_packages AS p + LEFT JOIN ossPackages_enriched_pkg_repo AS pr ON p.id = pr.packageId + LEFT JOIN ossPackages_enriched_repos AS r ON pr.repoId = r.repoId + LEFT JOIN ossPackages_enriched_snapshot AS snap ON pr.repoId = snap.repoId + LEFT JOIN ossPackages_enriched_maintainers AS mh ON p.id = mh.packageId + LEFT JOIN ossPackages_enriched_release AS rel ON p.id = rel.packageId + LEFT JOIN ossPackages_enriched_open_vulns AS ov ON p.id = ov.packageId + LEFT JOIN ossPackages_enriched_dep_vulns AS dv ON p.id = dv.packageId + LEFT JOIN ossPackages_enriched_repo_orgs AS ro ON p.repositoryUrl = ro.url + +NODE ossPackages_enriched_result +SQL > + SELECT + * EXCEPT ( + busFactor, + orgDiversity, + responsiveness, + openVulns, + scorecardSub, + dependencyHealth, + securityPractices, + supplyChainIntegrity, + releaseCadence, + commitActivity, + issueResolution, + prMergeHealth + ) + FROM ossPackages_enriched_scored + +TYPE COPY +TARGET_DATASOURCE ossPackages_enriched_ds +COPY_MODE replace +COPY_SCHEDULE 30 2 * * * diff --git a/services/libs/tinybird/pipes/ossPackages_enriched_sink.pipe b/services/libs/tinybird/pipes/ossPackages_enriched_sink.pipe new file mode 100644 index 0000000000..24aecd9cd5 --- /dev/null +++ b/services/libs/tinybird/pipes/ossPackages_enriched_sink.pipe @@ -0,0 +1,29 @@ +DESCRIPTION > + Sink pipe that exports per-package enrichment (health/lifecycle) fields back to Kafka so a + Kafka Connect JDBC sink can update the Postgres `packages` rows. Scoped to critical packages + (isCritical = 1), since enrichment is only computed/meaningful for the tracked set. Columns + are aliased to the snake_case Postgres column names so the JDBC sink maps field -> column 1:1. + Runs daily after `ossPackages_enriched.pipe` (the 02:30 UTC enrichment COPY) has refreshed + `ossPackages_enriched_ds`. + +NODE ossPackages_enriched_sink_select +SQL > + SELECT + purl, + lifecycleLabel AS lifecycle_label, + healthScore AS health_score, + healthLabel AS health_label, + maintainerHealthScore AS maintainer_health_score, + securitySupplyChainScore AS security_supply_chain_score, + developmentActivityScore AS development_activity_score, + signalCoverageHealth AS signal_coverage_health + FROM ossPackages_enriched_ds + WHERE isCritical = 1 + +TYPE SINK +EXPORT_SERVICE kafka +EXPORT_CONNECTION_NAME lfx-oracle-kafka-streaming +EXPORT_SCHEDULE 0 4 * * * +EXPORT_FORMAT json +EXPORT_STRATEGY @new +EXPORT_KAFKA_TOPIC ossPackages_enriched_sink