1148 improve performance of npm package ingester (#1149)

NPM package ingester is more performent. avoid cases where a deadlock
can happen.

Author: breadchris

lunatrace/bsl/ingest-worker/cmd/ingestworker/vulnerability/vulnerability.go

@@ -101,14 +101,17 @@ func NewCommand(p Params) clifx.CommandResult {
 					Usage: "[file or directory]",
 					Flags: []cli.Flag{
 						&cli.StringFlag{
-							Name:     "source",
-							Usage:    "Where the vulnerabilities have been sourced from.",
-							Required: true,
+							Name:  "source",
+							Usage: "Where the vulnerabilities have been sourced from.",
 						},
 						&cli.StringFlag{
 							Name:  "source-relative-path",
 							Usage: "Relative path from within the source to where advisories are located.",
 						},
+						&cli.StringFlag{
+							Name:  "vuln-id",
+							Usage: "ID of a vulnerability to make sure all related information is ingested (ex. affected packages).",
+						},
 						&cli.BoolFlag{
 							Name:  "ingest-affected",
 							Usage: "Ensure for every affected package that all metadata is collected.",
@@ -119,9 +122,17 @@ func NewCommand(p Params) clifx.CommandResult {
 						advisoryLocation := ctx.Args().First()
 
 						source := ctx.String("source")
+						vulnID := ctx.String("vuln-id")
 						sourceRelativePath := ctx.String("source-relative-path")
 						ingestAffected := ctx.Bool("ingest-affected")
 
+						if vulnID != "" {
+							log.Info().
+								Str("vulnerability", vulnID).
+								Msg("processing vulnerability")
+							return p.AffectedIngester.Ingest(ctx.Context, vulnID)
+						}
+
 						log.Info().
 							Str("source", source).
 							Msg("starting vulnerability ingestion")

lunatrace/bsl/ingest-worker/pkg/metadata/ingester/sql.go

@@ -163,11 +163,13 @@ func (s *packageSqlIngester) mapMaintainers(ctx context.Context, packageId uuid.
 }
 
 func (s *packageSqlIngester) mapMaintainer(ctx context.Context, pm metadata.Maintainer) (uuid.UUID, error) {
-	return upsertMaintainer(ctx, s.deps.DB, model.Maintainer{
+	maintainer := model.Maintainer{
 		PackageManager: mapper.NpmV,
 		Email:          pm.Email,
 		Name:           util.Ptr(pm.Name),
-	})
+	}
+
+	return upsertMaintainer(ctx, s.deps.DB, maintainer)
 }
 
 func (s *packageSqlIngester) Ingest(ctx context.Context, pkg *metadata.PackageMetadata) (string, error) {

lunatrace/bsl/ingest-worker/pkg/metadata/ingester/sqlstms.go

@@ -22,6 +22,26 @@ import (
 )
 
 func upsertPackage(ctx context.Context, db *sql.DB, p model.Package) (id uuid.UUID, err error) {
+	selectReleaseDependencyPackage := Package.SELECT(
+		Package.Name,
+		Package.PackageManager,
+		Package.CustomRegistry,
+		Package.Description,
+	).WHERE(
+		Package.Name.EQ(postgres.String(p.Name)).
+			AND(Package.PackageManager.EQ(postgres.NewEnumValue(string(p.PackageManager)))).
+			AND(Package.CustomRegistry.EQ(postgres.String(p.CustomRegistry))),
+	)
+
+	var releaseDependencyPackage model.Package
+	err = selectReleaseDependencyPackage.QueryContext(ctx, db, &releaseDependencyPackage)
+	if err == nil {
+		// If the release dependency package already exists, we don't need to do anything
+		if releaseDependencyPackage.Description == p.Description {
+			return releaseDependencyPackage.ID, nil
+		}
+	}
+
 	packageInsert := Package.INSERT(
 		Package.Name,
 		Package.PackageManager,
@@ -48,6 +68,31 @@ func upsertPackage(ctx context.Context, db *sql.DB, p model.Package) (id uuid.UU
 }
 
 func upsertRelease(ctx context.Context, db *sql.DB, r model.Release) (id uuid.UUID, err error) {
+	selectRelease := Release.SELECT(
+		Release.AllColumns,
+	).WHERE(
+		Release.PackageID.EQ(postgres.UUID(r.PackageID)).
+			AND(Release.Version.EQ(postgres.String(r.Version))),
+	)
+
+	var release model.Release
+	err = selectRelease.QueryContext(ctx, db, &release)
+	if err == nil {
+		// TODO (cthompson) not easy to compare jsonb, so we're skipping this for now
+		// release.UpstreamData == r.UpstreamData &&
+
+		// If the release already exists, we don't need to do anything
+		if release.PublishingMaintainerID == r.PublishingMaintainerID &&
+			release.ReleaseTime == r.ReleaseTime &&
+			release.BlobHash == r.BlobHash &&
+			release.UpstreamBlobURL == r.UpstreamBlobURL {
+
+			// TODO (cthompson) update FetchedTime
+
+			return release.ID, nil
+		}
+	}
+
 	releaseInsert := Release.INSERT(
 		Release.PackageID,
 		Release.PublishingMaintainerID,
@@ -82,6 +127,23 @@ func upsertRelease(ctx context.Context, db *sql.DB, r model.Release) (id uuid.UU
 }
 
 func upsertReleaseDependencyPackage(ctx context.Context, db *sql.DB, p model.Package) (id uuid.UUID, err error) {
+	selectReleaseDependencyPackage := Package.SELECT(
+		Package.Name,
+		Package.PackageManager,
+		Package.CustomRegistry,
+	).WHERE(
+		Package.Name.EQ(postgres.String(p.Name)).
+			AND(Package.PackageManager.EQ(postgres.NewEnumValue(string(p.PackageManager)))).
+			AND(Package.CustomRegistry.EQ(postgres.String(p.CustomRegistry))),
+	)
+
+	var releaseDependencyPackage model.Package
+	err = selectReleaseDependencyPackage.QueryContext(ctx, db, &releaseDependencyPackage)
+	if err == nil {
+		// If the release dependency package already exists, we don't need to do anything
+		return releaseDependencyPackage.ID, nil
+	}
+
 	insertPackage := Package.INSERT(
 		Package.Name,
 		Package.PackageManager,
@@ -105,6 +167,26 @@ func upsertReleaseDependencyPackage(ctx context.Context, db *sql.DB, p model.Pac
 }
 
 func upsertReleaseDependency(ctx context.Context, db *sql.DB, r model.ReleaseDependency) (id uuid.UUID, err error) {
+	selectReleaseDependency := ReleaseDependency.SELECT(
+		ReleaseDependency.ID,
+		ReleaseDependency.ReleaseID,
+		ReleaseDependency.IsDev,
+	).WHERE(
+		ReleaseDependency.ReleaseID.EQ(postgres.UUID(r.ReleaseID)).
+			AND(ReleaseDependency.PackageName.EQ(postgres.String(r.PackageName))).
+			AND(ReleaseDependency.PackageVersionQuery.EQ(postgres.String(r.PackageVersionQuery))),
+	)
+
+	var releaseDependency model.ReleaseDependency
+	err = selectReleaseDependency.QueryContext(ctx, db, &releaseDependency)
+	if err == nil {
+		// If the release dependency already exists, we don't need to do anything
+		if releaseDependency.ReleaseID == r.ReleaseID &&
+			releaseDependency.IsDev == r.IsDev {
+			return releaseDependency.ID, nil
+		}
+	}
+
 	insertReleaseDependency := ReleaseDependency.INSERT(
 		ReleaseDependency.ReleaseID,
 		ReleaseDependency.DependencyPackageID,
@@ -131,18 +213,52 @@ func upsertReleaseDependency(ctx context.Context, db *sql.DB, r model.ReleaseDep
 }
 
 func upsertPackageMaintainer(ctx context.Context, db *sql.DB, p model.PackageMaintainer) error {
+	selectPackageMaintainer := PackageMaintainer.SELECT(
+		PackageMaintainer.AllColumns,
+	).WHERE(
+		PackageMaintainer.PackageID.EQ(postgres.UUID(p.PackageID)).
+			AND(PackageMaintainer.MaintainerID.EQ(postgres.UUID(p.MaintainerID))),
+	)
+
+	var packageMaintainer model.PackageMaintainer
+	err := selectPackageMaintainer.QueryContext(ctx, db, &packageMaintainer)
+	if err == nil {
+		// If the package maintainer already exists, we don't need to do anything
+		if packageMaintainer.PackageID == p.PackageID &&
+			packageMaintainer.MaintainerID == p.MaintainerID {
+			return nil
+		}
+	}
+
 	insertPackageMaintainer := PackageMaintainer.INSERT(
 		PackageMaintainer.PackageID,
 		PackageMaintainer.MaintainerID,
 	).MODEL(p).
 		ON_CONFLICT().
 		ON_CONSTRAINT("package_maintainer_package_id_maintainer_id_idx").
 		DO_NOTHING()
-	_, err := insertPackageMaintainer.ExecContext(ctx, db)
+	_, err = insertPackageMaintainer.ExecContext(ctx, db)
 	return err
 }
 
 func upsertMaintainer(ctx context.Context, db *sql.DB, m model.Maintainer) (id uuid.UUID, err error) {
+	selectMaintainer := Maintainer.SELECT(
+		Maintainer.AllColumns,
+	).WHERE(
+		Maintainer.Email.EQ(postgres.String(m.Email)),
+	)
+
+	var maintainer model.Maintainer
+	err = selectMaintainer.QueryContext(ctx, db, &maintainer)
+	if err == nil {
+		// If the maintainer already exists, we don't need to do anything
+		if maintainer.Email == m.Email &&
+			maintainer.Name == m.Name &&
+			maintainer.PackageManager == m.PackageManager {
+			return maintainer.ID, nil
+		}
+	}
+
 	insertMaintainer := Maintainer.INSERT(
 		Maintainer.Email,
 		Maintainer.Name,

lunatrace/bsl/ingest-worker/pkg/metadata/replicator/npm/api.go

@@ -172,7 +172,8 @@ func (s *npmAPIReplicator) ReplicatePackages(packages []string, resolvePackage b
 			table.PackageDownloadCount.Downloads,
 			table.PackageDownloadCount.PackageID,
 		).MODELS(packageDownloadCounts).
-			ON_CONFLICT(table.PackageDownloadCount.Name, table.PackageDownloadCount.Day).
+			ON_CONFLICT().
+			ON_CONSTRAINT("package_download_count_name_day_key").
 			DO_UPDATE(
 				postgres.SET(
 					table.PackageDownloadCount.Downloads.SET(
@@ -278,11 +279,8 @@ func (s *npmAPIReplicator) ReplicateVersionDownloadCounts(packageName string, re
 		table.PackageVersionDownloadCount.Downloads,
 		table.PackageVersionDownloadCount.Day,
 	).MODELS(models).
-		ON_CONFLICT(
-			table.PackageVersionDownloadCount.Name,
-			table.PackageVersionDownloadCount.Version,
-			table.PackageVersionDownloadCount.Day,
-		).
+		ON_CONFLICT().
+		ON_CONSTRAINT("package_version_download_count_name_version_day_key").
 		DO_UPDATE(
 			postgres.SET(
 				table.PackageVersionDownloadCount.Downloads.SET(