Skip to content

Commit 01cc843

Browse files
joanagmaiaCopilot
joanagmaia
and
Copilot
authored
feat: add Tinybird datasources for packages-db tables (CM-1219) (#4180)
Signed-off-by: Joana Maia <jmaia@contractor.linuxfoundation.org> Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
1 parent 920d0dc commit 01cc843

15 files changed

Lines changed: 572 additions & 11 deletions

‎backend/src/osspckgs/migrations/V1781009234__sequin_publication_and_rank_packages_sync.sql‎

Lines changed: 165 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,165 @@
1+
-- Wire packages-db into the Sequin → Kafka → Tinybird pipeline.
2+
--
3+
-- Two related changes bundled here because both serve the same goal — making
4+
-- packages-db row changes replicate cleanly into Tinybird:
5+
--
6+
-- 1. Publication + REPLICA IDENTITY FULL on the 11 tables the Tinybird
7+
-- datasources read from. publish_via_partition_root collapses the
8+
-- versions (32) / package_dependencies (64) partition leaves into a
9+
-- single logical topic each. REPLICA IDENTITY on a partitioned root
10+
-- does not cascade, so every leaf is set explicitly via pg_inherits.
11+
--
12+
-- 2. rank_packages() bumps last_synced_at on every UPDATE that touches a
13+
-- DS-exported field (impact, is_critical, last_rank_pass_at).
14+
-- last_synced_at is the Tinybird ENGINE_VER for the packages datasource;
15+
-- without this bump, ReplacingMergeTree may keep an older row when
16+
-- criticality changes without any other write path touching the row.
17+
18+
-- ─── 1. Sequin publication ──────────────────────────────────────────────────
19+
20+
DO $$
21+
BEGIN
22+
IF NOT EXISTS (
23+
SELECT 1 FROM pg_publication WHERE pubname = 'sequin_pub'
24+
) THEN
25+
CREATE PUBLICATION sequin_pub
26+
FOR TABLE
27+
packages,
28+
versions,
29+
package_dependencies,
30+
package_maintainers,
31+
package_repos,
32+
maintainers,
33+
repos,
34+
repo_scorecard_checks,
35+
advisories,
36+
advisory_packages,
37+
advisory_affected_ranges
38+
WITH (publish_via_partition_root = true);
39+
END IF;
40+
END$$;
41+
42+
ALTER TABLE public.packages REPLICA IDENTITY FULL;
43+
ALTER TABLE public.versions REPLICA IDENTITY FULL;
44+
ALTER TABLE public.package_dependencies REPLICA IDENTITY FULL;
45+
ALTER TABLE public.package_maintainers REPLICA IDENTITY FULL;
46+
ALTER TABLE public.package_repos REPLICA IDENTITY FULL;
47+
ALTER TABLE public.maintainers REPLICA IDENTITY FULL;
48+
ALTER TABLE public.repos REPLICA IDENTITY FULL;
49+
ALTER TABLE public.repo_scorecard_checks REPLICA IDENTITY FULL;
50+
ALTER TABLE public.advisories REPLICA IDENTITY FULL;
51+
ALTER TABLE public.advisory_packages REPLICA IDENTITY FULL;
52+
ALTER TABLE public.advisory_affected_ranges REPLICA IDENTITY FULL;
53+
54+
-- versions (32) and package_dependencies (64) are hash-partitioned. REPLICA
55+
-- IDENTITY on the partitioned root does not cascade; set it on every leaf.
56+
DO $$
57+
DECLARE
58+
parent_table text;
59+
partition_oid regclass;
60+
BEGIN
61+
FOREACH parent_table IN ARRAY ARRAY['public.versions', 'public.package_dependencies']
62+
LOOP
63+
FOR partition_oid IN
64+
SELECT inhrelid::regclass
65+
FROM pg_inherits
66+
WHERE inhparent = parent_table::regclass
67+
LOOP
68+
EXECUTE format('ALTER TABLE %s REPLICA IDENTITY FULL', partition_oid);
69+
END LOOP;
70+
END LOOP;
71+
END$$;
72+
73+
-- ─── 2. rank_packages() bumps last_synced_at ────────────────────────────────
74+
75+
CREATE OR REPLACE FUNCTION rank_packages(
76+
weight_downloads numeric DEFAULT 0.25,
77+
weight_dependent_packages numeric DEFAULT 0.25,
78+
weight_transitive numeric DEFAULT 0.50,
79+
critical_top_n_by_ecosystem jsonb DEFAULT '{"npm":400000,"go":100000,"maven":200000,"pypi":100000,"nuget":50000,"cargo":75000}'::jsonb
80+
)
81+
RETURNS TABLE(scored_rows int, ranked_rows int)
82+
LANGUAGE plpgsql AS $$
83+
DECLARE
84+
n_scored int;
85+
n_ranked int;
86+
BEGIN
87+
-- Step 1: score
88+
WITH percentile_scores AS (
89+
SELECT
90+
id,
91+
(
92+
weight_downloads * PERCENT_RANK() OVER (
93+
PARTITION BY ecosystem ORDER BY LOG(1 + COALESCE(downloads_last_30d, 0)))
94+
95+
+ weight_dependent_packages * PERCENT_RANK() OVER (
96+
PARTITION BY ecosystem ORDER BY LOG(1 + COALESCE(dependent_count, 0)))
97+
98+
+ weight_transitive * PERCENT_RANK() OVER (
99+
PARTITION BY ecosystem ORDER BY LOG(1 + COALESCE(transitive_dependent_count, 0)))
100+
)::numeric(10, 4) AS new_impact
101+
FROM packages
102+
WHERE ecosystem IN (SELECT jsonb_object_keys(critical_top_n_by_ecosystem))
103+
)
104+
UPDATE packages p
105+
SET impact = ps.new_impact,
106+
last_synced_at = NOW()
107+
FROM percentile_scores ps
108+
WHERE p.id = ps.id
109+
AND p.impact IS DISTINCT FROM ps.new_impact;
110+
111+
GET DIAGNOSTICS n_scored = ROW_COUNT;
112+
113+
-- Step 2: rank + flag
114+
WITH ranked AS (
115+
SELECT
116+
id, ecosystem,
117+
ROW_NUMBER() OVER (
118+
PARTITION BY ecosystem
119+
ORDER BY impact DESC NULLS LAST, id
120+
) AS r
121+
FROM packages
122+
WHERE purl IS NOT NULL
123+
AND ecosystem IN (SELECT jsonb_object_keys(critical_top_n_by_ecosystem))
124+
),
125+
flagged AS (
126+
SELECT
127+
id, r,
128+
COALESCE(
129+
r <= (critical_top_n_by_ecosystem ->> ecosystem)::int,
130+
FALSE
131+
) AS new_is_critical
132+
FROM ranked
133+
)
134+
UPDATE packages p
135+
SET rank_in_ecosystem = f.r,
136+
is_critical = f.new_is_critical,
137+
last_synced_at = NOW()
138+
FROM flagged f
139+
WHERE p.id = f.id
140+
AND (
141+
p.rank_in_ecosystem IS DISTINCT FROM f.r
142+
OR p.is_critical IS DISTINCT FROM f.new_is_critical
143+
);
144+
145+
GET DIAGNOSTICS n_ranked = ROW_COUNT;
146+
147+
-- Step 2.5: spotlight overrides
148+
UPDATE packages p
149+
SET is_critical = TRUE,
150+
last_synced_at = NOW()
151+
FROM package_criticality_spotlight s
152+
WHERE p.ecosystem = s.ecosystem
153+
AND (p.namespace IS NOT DISTINCT FROM s.namespace)
154+
AND p.name = s.name
155+
AND p.is_critical = FALSE;
156+
157+
-- Step 3: stamp last_rank_pass_at unconditionally
158+
UPDATE packages
159+
SET last_rank_pass_at = NOW(),
160+
last_synced_at = NOW()
161+
WHERE ecosystem IN (SELECT jsonb_object_keys(critical_top_n_by_ecosystem));
162+
163+
RETURN QUERY SELECT n_scored, n_ranked;
164+
END;
165+
$$;

‎services/apps/packages_worker/src/deps-dev/workflows/ingestRepos.ts‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -58,11 +58,13 @@ const REPOS_PG_COLUMNS = [
5858
'open_issues',
5959
]
6060

61+
// last_synced_at is intentionally left NULL on seed — it is owned by the GitHub
62+
// enricher as its freshness signal. created_at / updated_at use their column defaults.
6163
const REPOS_MERGE_SQL = `
6264
INSERT INTO repos (url, raw_project_type, raw_project_name, host, owner, name,
63-
description, homepage, stars, forks, open_issues, last_synced_at)
65+
description, homepage, stars, forks, open_issues)
6466
SELECT s.canonical_url, s.raw_project_type, s.raw_project_name, s.host, s.owner, s.name,
65-
s.description, s.homepage, s.stars, s.forks, s.open_issues, NOW()
67+
s.description, s.homepage, s.stars, s.forks, s.open_issues
6668
FROM staging.osspckgs_repos_raw s
6769
ON CONFLICT (url) DO NOTHING
6870
`

‎services/libs/data-access-layer/src/osspckgs/repos.ts‎

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -15,19 +15,20 @@ export async function findRepoIdsByUrl(
1515
* Inserts or updates a repo row keyed on url.
1616
* Uses COALESCE so richer data from other enrichers (GitHub, deps.dev) is never
1717
* overwritten with nulls from a partial write.
18+
* `last_synced_at` is intentionally NOT touched here — that column is owned by the
19+
* GitHub enricher as its freshness signal. Maven discovery only stamps updated_at.
1820
* Returns the repo id.
1921
*/
2022
export async function upsertRepo(qx: QueryExecutor, item: IDbRepoUpsert): Promise<number> {
2123
const row = await qx.selectOne(
2224
`
23-
INSERT INTO repos (url, host, owner, name, last_synced_at, updated_at)
24-
VALUES ($(url), $(host), $(owner), $(name), NOW(), NOW())
25+
INSERT INTO repos (url, host, owner, name, updated_at)
26+
VALUES ($(url), $(host), $(owner), $(name), NOW())
2527
ON CONFLICT (url) DO UPDATE SET
26-
host = COALESCE(EXCLUDED.host, repos.host),
27-
owner = COALESCE(EXCLUDED.owner, repos.owner),
28-
name = COALESCE(EXCLUDED.name, repos.name),
29-
last_synced_at = NOW(),
30-
updated_at = NOW()
28+
host = COALESCE(EXCLUDED.host, repos.host),
29+
owner = COALESCE(EXCLUDED.owner, repos.owner),
30+
name = COALESCE(EXCLUDED.name, repos.name),
31+
updated_at = NOW()
3132
RETURNING id
3233
`,
3334
item,

‎services/libs/data-access-layer/src/packages/osv.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -243,7 +243,7 @@ export async function getRangesForPackages(qx: QueryExecutor, ids: number[]): Pr
243243
export async function flipVulnerableFlags(qx: QueryExecutor, ids: number[]): Promise<number> {
244244
if (ids.length === 0) return 0
245245
return qx.result(
246-
`UPDATE packages SET has_critical_vulnerability = TRUE
246+
`UPDATE packages SET has_critical_vulnerability = TRUE, last_synced_at = NOW()
247247
WHERE id IN ($(ids:csv)) AND has_critical_vulnerability = FALSE`,
248248
{ ids },
249249
)
@@ -252,7 +252,7 @@ export async function flipVulnerableFlags(qx: QueryExecutor, ids: number[]): Pro
252252
export async function clearSafeFlags(qx: QueryExecutor, ids: number[]): Promise<number> {
253253
if (ids.length === 0) return 0
254254
return qx.result(
255-
`UPDATE packages SET has_critical_vulnerability = FALSE
255+
`UPDATE packages SET has_critical_vulnerability = FALSE, last_synced_at = NOW()
256256
WHERE id IN ($(ids:csv)) AND has_critical_vulnerability = TRUE`,
257257
{ ids },
258258
)

‎services/libs/tinybird/datasources/advisories.datasource‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
DESCRIPTION >
2+
- `advisories` contains OSV-shaped security advisories ingested from deps.dev BigQuery.
3+
- Replicated from Postgres packages-db — each row is one advisory (CVE, GHSA, MAL-*, etc.).
4+
- Used to surface critical vulnerabilities, track CVSS scores, and power security overlays in LFX Insights.
5+
- `id` is the internal primary key.
6+
- `osvId` is the canonical advisory identifier (e.g. GHSA-xxx, CVE-xxx, OSV-xxx) — globally unique.
7+
- `source` is the originating database: 'GHSA', 'OSV', 'NVD', 'NSWG', etc. (empty string if unknown).
8+
- `sourceUrl` is the upstream advisory URL (empty string if not provided).
9+
- `aliases` is an array of alternate identifiers for the same advisory (e.g. a GHSA may alias a CVE).
10+
- `severity` is the qualitative risk level: 'LOW', 'MEDIUM', 'HIGH', or 'CRITICAL' (empty string if unknown).
11+
- `cvss` is the numeric CVSS score (0 if not available).
12+
- `cvssSource` documents the provenance of the CVSS value: 'osv_cvss_v3', 'osv_qualitative_fallback', 'osv_malicious_package', etc.
13+
- `isCritical` is 1 when cvss >= 7.0 (HIGH or CRITICAL), 0 otherwise — computed from the score.
14+
- `summary` is a short human-readable description of the vulnerability (empty string if not provided).
15+
- `details` is the full advisory text (empty string if not provided).
16+
- `publishedAt` is when the advisory was first published upstream (NULL if unknown).
17+
- `modifiedAt` is when the advisory was last modified upstream; NULL for BQ-sourced rows.
18+
- `createdAt` and `updatedAt` are row-level audit timestamps for Tinybird watermark-based sync.
19+
20+
SCHEMA >
21+
`id` UInt64 `json:$.record.id`,
22+
`osvId` String `json:$.record.osv_id`,
23+
`source` String `json:$.record.source` DEFAULT '',
24+
`sourceUrl` String `json:$.record.source_url` DEFAULT '',
25+
`aliases` Array(String) `json:$.record.aliases[:]` DEFAULT [],
26+
`severity` String `json:$.record.severity` DEFAULT '',
27+
`cvss` Float32 `json:$.record.cvss` DEFAULT 0,
28+
`cvssSource` String `json:$.record.cvss_source` DEFAULT '',
29+
`isCritical` UInt8 `json:$.record.is_critical` DEFAULT 0,
30+
`summary` String `json:$.record.summary` DEFAULT '',
31+
`details` String `json:$.record.details` DEFAULT '',
32+
`publishedAt` Nullable(DateTime64(3)) `json:$.record.published_at`,
33+
`modifiedAt` Nullable(DateTime64(3)) `json:$.record.modified_at`,
34+
`createdAt` DateTime64(3) `json:$.record.created_at`,
35+
`updatedAt` DateTime64(3) `json:$.record.updated_at`
36+
37+
ENGINE ReplacingMergeTree
38+
ENGINE_PARTITION_KEY toYear(createdAt)
39+
ENGINE_SORTING_KEY osvId
40+
ENGINE_VER updatedAt

‎services/libs/tinybird/datasources/advisoryAffectedRanges.datasource‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
DESCRIPTION >
2+
- `advisoryAffectedRanges` stores the version ranges affected by a security advisory per package.
3+
- Replicated from Postgres packages-db — each row defines one vulnerable window (introduced → fixed/last_affected).
4+
- Used to determine whether a specific package version falls within a known vulnerable range.
5+
- `id` is the internal primary key.
6+
- `advisoryPackageId` links to the parent advisory_packages row.
7+
- `introducedVersion` is the version where the vulnerability was introduced (empty string if unknown start).
8+
- `fixedVersion` is the version where the vulnerability was patched (empty string if no fix exists yet).
9+
- `lastAffected` is the last known affected version when there is no fixed version (empty string if not applicable).
10+
- `rangeRaw` is the raw AffectedVersions string from the deps.dev BigQuery source (empty string if OSV-sourced).
11+
- `unaffectedRaw` is the raw UnaffectedVersions string from deps.dev BigQuery (empty string if OSV-sourced).
12+
- `createdAt` and `updatedAt` are row-level audit timestamps for Tinybird watermark-based sync.
13+
14+
SCHEMA >
15+
`id` UInt64 `json:$.record.id`,
16+
`advisoryPackageId` UInt64 `json:$.record.advisory_package_id`,
17+
`introducedVersion` String `json:$.record.introduced_version` DEFAULT '',
18+
`fixedVersion` String `json:$.record.fixed_version` DEFAULT '',
19+
`lastAffected` String `json:$.record.last_affected` DEFAULT '',
20+
`rangeRaw` String `json:$.record.range_raw` DEFAULT '',
21+
`unaffectedRaw` String `json:$.record.unaffected_raw` DEFAULT '',
22+
`createdAt` DateTime64(3) `json:$.record.created_at`,
23+
`updatedAt` DateTime64(3) `json:$.record.updated_at`
24+
25+
ENGINE ReplacingMergeTree
26+
ENGINE_PARTITION_KEY toYear(createdAt)
27+
ENGINE_SORTING_KEY advisoryPackageId, id
28+
ENGINE_VER updatedAt

‎services/libs/tinybird/datasources/advisoryPackages.datasource‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
DESCRIPTION >
2+
- `advisoryPackages` maps security advisories to the packages they affect.
3+
- Replicated from Postgres packages-db — one advisory can affect multiple packages across different ecosystems.
4+
- Used to determine which packages are impacted by a given advisory, and to backfill `has_critical_vulnerability` on packages.
5+
- `id` is the internal primary key.
6+
- `advisoryId` links to the parent advisory record.
7+
- `packageId` links to the matching packages row (0 if the package exists in OSV but not yet in our DB).
8+
- `ecosystem` is the package ecosystem (npm, go, maven, pypi, etc.).
9+
- `packageName` is the package name within its ecosystem as reported by OSV.
10+
- `createdAt` and `updatedAt` are row-level audit timestamps for Tinybird watermark-based sync.
11+
12+
SCHEMA >
13+
`id` UInt64 `json:$.record.id`,
14+
`advisoryId` UInt64 `json:$.record.advisory_id`,
15+
`packageId` UInt64 `json:$.record.package_id` DEFAULT 0,
16+
`ecosystem` String `json:$.record.ecosystem`,
17+
`packageName` String `json:$.record.package_name`,
18+
`createdAt` DateTime64(3) `json:$.record.created_at`,
19+
`updatedAt` DateTime64(3) `json:$.record.updated_at`
20+
21+
ENGINE ReplacingMergeTree
22+
ENGINE_PARTITION_KEY toYear(createdAt)
23+
ENGINE_SORTING_KEY ecosystem, packageName, id
24+
ENGINE_VER updatedAt

‎services/libs/tinybird/datasources/maintainers.datasource‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
DESCRIPTION >
2+
- `maintainers` contains package maintainer profiles sourced from package registries (npm, PyPI, etc.).
3+
- Replicated from Postgres packages-db — one row per unique (ecosystem, username) identity.
4+
- Used to identify who maintains critical packages, correlate maintainers across ecosystems, and support contributor analytics.
5+
- `id` is the internal primary key.
6+
- `ecosystem` is the package registry this identity belongs to: 'npm', 'pypi', 'cargo', 'maven', etc.
7+
- `username` is the maintainer's registry username — unique within an ecosystem.
8+
- `displayName` is the maintainer's human-readable name as published in the registry (empty string if not provided).
9+
- `url` is the maintainer's profile URL on the registry (empty string if not provided).
10+
- `email` is the maintainer's email address as published in the registry (empty string if not provided).
11+
- `githubLogin` is the maintainer's GitHub username if resolved (empty string if not linked).
12+
- `createdAt` and `updatedAt` are row-level audit timestamps for Tinybird watermark-based sync.
13+
14+
SCHEMA >
15+
`id` UInt64 `json:$.record.id`,
16+
`ecosystem` String `json:$.record.ecosystem`,
17+
`username` String `json:$.record.username`,
18+
`displayName` String `json:$.record.display_name` DEFAULT '',
19+
`url` String `json:$.record.url` DEFAULT '',
20+
`email` String `json:$.record.email` DEFAULT '',
21+
`githubLogin` String `json:$.record.github_login` DEFAULT '',
22+
`createdAt` DateTime64(3) `json:$.record.created_at`,
23+
`updatedAt` DateTime64(3) `json:$.record.updated_at`
24+
25+
ENGINE ReplacingMergeTree
26+
ENGINE_PARTITION_KEY toYear(createdAt)
27+
ENGINE_SORTING_KEY ecosystem, username
28+
ENGINE_VER updatedAt

‎services/libs/tinybird/datasources/packageDependencies.datasource‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
DESCRIPTION >
2+
- `packageDependencies` stores the dependency graph between package versions.
3+
- Replicated from Postgres packages-db — one row per (version, dependency) edge in the dependency graph.
4+
- Partitioned by hash(depends_on_id) in Postgres for fast downstream lookups; sorted here for analytical queries.
5+
- Used to answer "what depends on package X?" (downstream consumers) and to compute dependent_repos_count / dependent_packages_count.
6+
- `id` is the internal primary key.
7+
- `packageId` is the ID of the package that contains the depending version.
8+
- `versionId` is the specific version that declares the dependency.
9+
- `dependsOnId` is the package being depended upon — the hot lookup key for vulnerability blast-radius queries.
10+
- `dependsOnVersionId` is the resolved specific version of the dependency (0 if the exact version is unknown).
11+
- `versionConstraint` is the declared version constraint string (e.g. '^1.2.3', '>=2.0.0'); empty string if not specified.
12+
- `dependencyKind` is the dependency type: 'direct', 'dev', or 'peer'.
13+
- `isOptional` is 1 if the dependency is marked optional, 0 otherwise.
14+
- `createdAt` and `updatedAt` are row-level audit timestamps for Tinybird watermark-based sync.
15+
16+
SCHEMA >
17+
`id` UInt64 `json:$.record.id`,
18+
`packageId` UInt64 `json:$.record.package_id`,
19+
`versionId` UInt64 `json:$.record.version_id`,
20+
`dependsOnId` UInt64 `json:$.record.depends_on_id`,
21+
`dependsOnVersionId` UInt64 `json:$.record.depends_on_version_id` DEFAULT 0,
22+
`versionConstraint` String `json:$.record.version_constraint` DEFAULT '',
23+
`dependencyKind` String `json:$.record.dependency_kind`,
24+
`isOptional` UInt8 `json:$.record.is_optional` DEFAULT 0,
25+
`createdAt` DateTime64(3) `json:$.record.created_at`,
26+
`updatedAt` DateTime64(3) `json:$.record.updated_at`
27+
28+
ENGINE ReplacingMergeTree
29+
ENGINE_PARTITION_KEY toYear(createdAt)
30+
ENGINE_SORTING_KEY dependsOnId, versionId, id
31+
ENGINE_VER updatedAt

0 commit comments

Comments
 (0)