adds streaming transformers, streaming gob files straight to database

.github/workflows/vulndb.yaml

@@ -22,7 +22,7 @@ jobs:
       FRONTEND_URL: "doesntmatter"
     services:
       postgres:
-        image: ghcr.io/l3montree-dev/devguard-postgresql:v0.5.3@sha256:a06c9e7c8ee334790cc66d52e89ff5ef05352ab264841d3d9f3659c046732251
+        image: ghcr.io/l3montree-dev/devguard/postgresql:v1.3.1
         env:
           POSTGRES_DB: ${{env.POSTGRES_DB}}
           POSTGRES_USER: ${{env.POSTGRES_USER}}

tests/db_init.go

@@ -4,11 +4,14 @@ import (
 	"context"
 	"log"
 	"log/slog"
+	"os"
+	"path/filepath"
 	"time"
 
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/l3montree-dev/devguard/database"
 	"github.com/l3montree-dev/devguard/shared"
+	"github.com/moby/moby/api/types/container"
 	"github.com/testcontainers/testcontainers-go"
 	"github.com/testcontainers/testcontainers-go/modules/postgres"
 )
@@ -34,13 +37,40 @@ func InitRawDatabaseContainer(initDBSQLPath string) (*pgxpool.Pool, func()) {
 	dbUser := "user"
 	dbPassword := "password"
 
+	// The image has a read-only Nix filesystem so docker cp (used by WithInitScripts)
+	// cannot write into the container. Instead we bind-mount the init SQL file,
+	// mirroring how docker-compose.yaml mounts ./initdb.sql.
+	absInitSQL, err := filepath.Abs(initDBSQLPath)
+	if err != nil {
+		panic("could not resolve initdb SQL path: " + err.Error())
+	}
+	if _, err := os.Stat(absInitSQL); err != nil {
+		panic("initdb SQL file not found: " + absInitSQL)
+	}
+
 	postgresC, err := postgres.Run(ctx,
-		"ghcr.io/l3montree-dev/devguard-postgresql:v0.4.16",
+		"ghcr.io/l3montree-dev/devguard/postgresql:v1.3.1",
 		postgres.WithDatabase(dbName),
 		postgres.WithUsername(dbUser),
 		postgres.WithPassword(dbPassword),
-		postgres.WithInitScripts(initDBSQLPath),
 		postgres.BasicWaitStrategies(),
+		testcontainers.WithLogger(log.Default()),
+		// The postgres module overrides CMD to "postgres -c fsync=off", which drops the
+		// image's config_file arg and makes postgres listen only on 127.0.0.1. We restore
+		// the config_file so listen_addresses='*' takes effect for port mapping.
+		testcontainers.WithCmd("postgres",
+			"-c", "config_file=/etc/postgresql/postgresql.conf",
+			"-c", "fsync=off",
+		),
+		testcontainers.WithTmpfs(map[string]string{
+			"/run/postgresql": "rw",
+		}),
+		testcontainers.WithHostConfigModifier(func(hc *container.HostConfig) {
+			hc.ShmSize = 1 << 30 // 1 GiB — matches shm_size in docker-compose.yaml
+			// Bind-mount the init SQL; WithInitScripts uses docker cp which fails on the
+			// read-only Nix filesystem of this image.
+			hc.Binds = append(hc.Binds, absInitSQL+":/docker-entrypoint-initdb.d/init.sql:ro")
+		}),
 	)
 
 	terminate := func() {
@@ -49,10 +79,40 @@ func InitRawDatabaseContainer(initDBSQLPath string) (*pgxpool.Pool, func()) {
 		}
 	}
 	if err != nil {
+		if postgresC != nil {
+			if logs, lerr := postgresC.Logs(ctx); lerr == nil {
+				log.Printf("=== container logs ===")
+				buf := make([]byte, 64*1024)
+				for {
+					n, rerr := logs.Read(buf)
+					if n > 0 {
+						log.Printf("%s", buf[:n])
+					}
+					if rerr != nil {
+						break
+					}
+				}
+				logs.Close()
+			}
+		}
 		slog.Info("failed to start postgres container", "error", err)
 		panic(err)
 	}
 
+	if logs, lerr := postgresC.Logs(ctx); lerr == nil {
+		buf := make([]byte, 64*1024)
+		for {
+			n, rerr := logs.Read(buf)
+			if n > 0 {
+				log.Printf("=== postgres startup logs ===\n%s", buf[:n])
+			}
+			if rerr != nil {
+				break
+			}
+		}
+		logs.Close()
+	}
+
 	host, _ := postgresC.Host(ctx)
 	port, _ := postgresC.MappedPort(ctx, "5432")
 

tests/fx_test_helpers.go

@@ -31,7 +31,7 @@ import (
 
 // TestFixture provides a complete test environment with database and FX app
 type TestFixture struct {
-	T   *testing.T
+	T   testing.TB
 	App *TestApp
 	DB  shared.DB
 	// pool is the underlying pgx connection pool
@@ -40,7 +40,7 @@ type TestFixture struct {
 }
 
 // NewTestFixture creates a complete test environment with database container and FX app
-func NewTestFixture(t *testing.T, sqlInitFile string, options *TestAppOptions) *TestFixture {
+func NewTestFixture(t testing.TB, sqlInitFile string, options *TestAppOptions) *TestFixture {
 	t.Helper()
 
 	// Initialize database container

tests/vulndb_import_bench_test.go

@@ -0,0 +1,94 @@
+// Copyright (C) 2026 l3montree GmbH
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+package tests
+
+import (
+	"context"
+	"os"
+	"runtime"
+	"runtime/pprof"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// BenchmarkImportRC measures the full memory and time cost of VulnDBService.ImportRC,
+// including gob decoding and all database writes via a real PostgreSQL container.
+//
+// The benchmark writes a heap profile to mem.prof after each run.
+// View it as an interactive HTML flamegraph with:
+//
+//	go tool pprof -http=:8080 mem.prof
+//
+// Enable the debugImport const in vulndb/vulndb_service.go to cache vulndb.tar.zst
+// locally (in the working directory) so the archive is only downloaded once across
+// repeated runs.
+//
+// Run with:
+//
+//	go test -bench=BenchmarkImportRC -benchmem -run=^$ -timeout=30m ./tests/
+func BenchmarkImportRC(b *testing.B) {
+	ctx := context.Background()
+
+	fixture := NewTestFixture(b, "../initdb.sql", &TestAppOptions{SuppressLogs: true})
+
+	for b.Loop() {
+		var memBefore, memAfter runtime.MemStats
+		runtime.GC()
+		runtime.ReadMemStats(&memBefore)
+
+		var peakHeap atomic.Uint64
+		done := make(chan struct{})
+		go func() {
+			var ms runtime.MemStats
+			for {
+				select {
+				case <-done:
+					return
+				case <-time.After(100 * time.Millisecond):
+					runtime.ReadMemStats(&ms)
+					if ms.HeapInuse > peakHeap.Load() {
+						peakHeap.Store(ms.HeapInuse)
+					}
+				}
+			}
+		}()
+
+		if err := fixture.App.VulnDBService.ImportRC(ctx); err != nil {
+			close(done)
+			b.Fatalf("ImportRC failed: %v", err)
+		}
+		close(done)
+
+		runtime.GC()
+		runtime.ReadMemStats(&memAfter)
+
+		b.ReportMetric(float64(peakHeap.Load())/1024/1024, "peak_heap_MiB")
+		b.ReportMetric(float64(memAfter.HeapInuse-memBefore.HeapInuse)/1024/1024, "heap_MiB")
+		b.ReportMetric(float64(memAfter.TotalAlloc-memBefore.TotalAlloc)/1024/1024, "total_alloc_MiB")
+
+		// Write a heap profile after every iteration so the last (or only) run is always captured.
+		f, err := os.Create("mem.prof")
+		if err != nil {
+			b.Fatalf("could not create mem.prof: %v", err)
+		}
+		if err := pprof.WriteHeapProfile(f); err != nil {
+			f.Close()
+			b.Fatalf("could not write heap profile: %v", err)
+		}
+		f.Close()
+	}
+}

vulndb/exploitdb_service.go

@@ -105,24 +105,8 @@ func insertExploitsBulk(ctx context.Context, tx pgx.Tx, exploits []models.Exploi
 	if len(exploits) == 0 {
 		return nil
 	}
-	if _, err := tx.Exec(ctx, `
-		CREATE TEMP TABLE exploits_stage (
-			id          text,
-			published   date,
-			updated     date,
-			author      text,
-			type        text,
-			verified    boolean,
-			source_url  text,
-			description text,
-			cve_id      text,
-			tags        text,
-			forks       integer,
-			watchers    integer,
-			subscribers integer,
-			stars       integer
-		) ON COMMIT DROP`); err != nil {
-		return fmt.Errorf("could not create exploits staging table: %w", err)
+	if _, err := tx.Exec(ctx, `TRUNCATE exploits_stage`); err != nil {
+		return fmt.Errorf("could not truncate exploits staging table: %w", err)
 	}
 
 	if _, err := tx.CopyFrom(ctx, pgx.Identifier{"exploits_stage"},

vulndb/gob_types.go

@@ -52,7 +52,7 @@ type GobMaliciousComponent struct {
 // GobMaliciousPackagesExport bundles the full malicious-packages snapshot.
 // models.MaliciousPackage only contains plain types and is gob-safe directly.
 type GobMaliciousPackagesExport struct {
-	Packages   []models.MaliciousPackage
+	Package    models.MaliciousPackage
 	Components []GobMaliciousComponent
 }
 
@@ -120,17 +120,6 @@ func gobExploitToModel(g GobExploit) models.Exploit {
 	}
 }
 
-func gobExploitsToModels(gs []GobExploit, lastImportTime time.Time) []models.Exploit {
-	out := make([]models.Exploit, 0, len(gs))
-	for _, g := range gs {
-		if g.Updated != nil && g.Updated.Before(lastImportTime) {
-			continue
-		}
-		out = append(out, gobExploitToModel(g))
-	}
-	return out
-}
-
 // --- Malicious package conversions ---
 
 func maliciousComponentToGob(c models.MaliciousAffectedComponent) GobMaliciousComponent {
@@ -160,31 +149,3 @@ func gobComponentToModel(g GobMaliciousComponent) models.MaliciousAffectedCompon
 		VersionFixed:       g.VersionFixed,
 	}
 }
-
-func malPackagesExportToGob(packages []models.MaliciousPackage, components []models.MaliciousAffectedComponent) GobMaliciousPackagesExport {
-	gobComps := make([]GobMaliciousComponent, len(components))
-	for i, c := range components {
-		gobComps[i] = maliciousComponentToGob(c)
-	}
-	return GobMaliciousPackagesExport{Packages: packages, Components: gobComps}
-}
-
-func gobMalPackagesExportToModels(g GobMaliciousPackagesExport, lastImportTime time.Time) ([]models.MaliciousPackage, []models.MaliciousAffectedComponent) {
-	// build a map of package ID to last import time for all packages in the export
-	pkgImportTimes := make(map[string]struct{})
-	filteredPkgs := make([]models.MaliciousPackage, 0, len(g.Packages))
-	for _, pkg := range g.Packages {
-		if pkg.Modified.After(lastImportTime) {
-			pkgImportTimes[pkg.ID] = struct{}{}
-			filteredPkgs = append(filteredPkgs, pkg)
-		}
-	}
-	comps := make([]models.MaliciousAffectedComponent, 0, len(g.Components))
-	for _, c := range g.Components {
-		if _, ok := pkgImportTimes[c.MaliciousPackageID]; !ok {
-			continue
-		}
-		comps = append(comps, gobComponentToModel(c))
-	}
-	return filteredPkgs, comps
-}

vulndb/malicious_packages_checker.go

@@ -50,6 +50,12 @@ type MaliciousPackageChecker struct {
 	httpClient *http.Client
 }
 
+type malRow struct {
+	pkgs  []models.MaliciousPackage
+	comps []models.MaliciousAffectedComponent
+}
+
+
 func NewMaliciousPackageChecker(
 	repository *repositories.MaliciousPackageRepository,
 ) (*MaliciousPackageChecker, error) {
@@ -286,15 +292,8 @@ func (c *MaliciousPackageChecker) IsMalicious(ctx context.Context, ecosystem, pa
 
 func insertMaliciousPackagesBulk(ctx context.Context, tx pgx.Tx, pkgs []models.MaliciousPackage, comps []models.MaliciousAffectedComponent) error {
 	if len(pkgs) > 0 {
-		if _, err := tx.Exec(ctx, `
-			CREATE TEMP TABLE mal_pkgs_stage (
-				id        text,
-				summary   text,
-				details   text,
-				published timestamptz,
-				modified  timestamptz
-			) ON COMMIT DROP`); err != nil {
-			return fmt.Errorf("could not create malicious packages staging table: %w", err)
+		if _, err := tx.Exec(ctx, `TRUNCATE mal_pkgs_stage`); err != nil {
+			return fmt.Errorf("could not truncate malicious packages staging table: %w", err)
 		}
 		if _, err := tx.CopyFrom(ctx, pgx.Identifier{"mal_pkgs_stage"},
 			[]string{"id", "summary", "details", "published", "modified"},
@@ -317,19 +316,8 @@ func insertMaliciousPackagesBulk(ctx context.Context, tx pgx.Tx, pkgs []models.M
 	}
 
 	if len(comps) > 0 {
-		if _, err := tx.Exec(ctx, `
-			CREATE TEMP TABLE mal_comps_stage (
-				id                   text,
-				malicious_package_id text,
-				purl                 text,
-				ecosystem            text,
-				version              text,
-				semver_introduced    text,
-				semver_fixed         text,
-				version_introduced   text,
-				version_fixed        text
-			) ON COMMIT DROP`); err != nil {
-			return fmt.Errorf("could not create malicious components staging table: %w", err)
+		if _, err := tx.Exec(ctx, `TRUNCATE mal_comps_stage`); err != nil {
+			return fmt.Errorf("could not truncate malicious components staging table: %w", err)
 		}
 		if _, err := tx.CopyFrom(ctx, pgx.Identifier{"mal_comps_stage"},
 			[]string{"id", "malicious_package_id", "purl", "ecosystem", "version", "semver_introduced", "semver_fixed", "version_introduced", "version_fixed"},