internal/worker: insert module if not in module_version_states

If a module is introduced for the first time to pkgsite through frontend
fetch, and not the index, it won't have a row in module_version_states.

A row is now inserted before the fetch process begins if the proxy's
info endpoint tells us that this is a valid module.

This fixes a bug in the fetch flow where
module_version_states.last_processed_at is not updated in
upsertModuleVersionState if a row did not already exist.

For golang/go#46985

Change-Id: I2a166228df20b1fb935bbabebb5a651da0e2c1ba
Reviewed-on: https://go-review.googlesource.com/c/pkgsite/+/341892
Trust: Julie Qiu <julie@golang.org>
Run-TryBot: Julie Qiu <julie@golang.org>
TryBot-Result: kokoro <noreply+kokoro@google.com>
Reviewed-by: Jonathan Amsterdam <jba@google.com>

1 files changed, 44 insertions, 9 deletions

diff --git a/internal/postgres/versionstate.go b/internal/postgres/versionstate.go
index 6f043725..85f8d54a 100644
--- a/internal/postgres/versionstate.go
+++ b/internal/postgres/versionstate.go
@@ -24,19 +24,49 @@ import (
 // table with a status of zero.
 func (db *DB) InsertIndexVersions(ctx context.Context, versions []*internal.IndexVersion) (err error) {
 	defer derrors.WrapStack(&err, "InsertIndexVersions(ctx, %v)", versions)
-	var vals []interface{}
-	for _, v := range versions {
-		vals = append(vals, v.Path, v.Version, version.ForSorting(v.Version), v.Timestamp, 0, "", "", version.IsIncompatible(v.Version))
-	}
-	cols := []string{"module_path", "version", "sort_version", "index_timestamp", "status", "error", "go_mod_path", "incompatible"}
 	conflictAction := `
 		ON CONFLICT
 			(module_path, version)
 		DO UPDATE SET
 			index_timestamp=excluded.index_timestamp,
-			next_processed_after=CURRENT_TIMESTAMP
-	`
-	return db.db.Transact(ctx, sql.LevelDefault, func(tx *database.DB) error {
+			next_processed_after=CURRENT_TIMESTAMP`
+	return insertIndexVersions(ctx, db.db, versions, conflictAction)
+}
+
+// InsertNewModuleVersionFromFrontendFetch inserts a new module version into
+// the module_version_states table with a status of zero that was requested
+// from frontend fetch.
+func (db *DB) InsertNewModuleVersionFromFrontendFetch(ctx context.Context, modulePath, resolvedVersion string) (err error) {
+	defer derrors.WrapStack(&err, "InsertIndexVersion(ctx, %v)", resolvedVersion)
+	conflictAction := `ON CONFLICT (module_path, version) DO NOTHING`
+	return insertIndexVersions(ctx, db.db, []*internal.IndexVersion{{Path: modulePath, Version: resolvedVersion}}, conflictAction)
+}
+
+func insertIndexVersions(ctx context.Context, ddb *database.DB, versions []*internal.IndexVersion, conflictAction string) (err error) {
+	var vals []interface{}
+	for _, v := range versions {
+		vals = append(vals,
+			v.Path,
+			v.Version,
+			version.ForSorting(v.Version),
+			0,
+			"",
+			"",
+			version.IsIncompatible(v.Version),
+			v.Timestamp,
+		)
+	}
+	cols := []string{
+		"module_path",
+		"version",
+		"sort_version",
+		"status",
+		"error",
+		"go_mod_path",
+		"incompatible",
+		"index_timestamp",
+	}
+	return ddb.Transact(ctx, sql.LevelDefault, func(tx *database.DB) error {
 		var updates [][2]string // (module_path, version) to update status
 		err := tx.BulkInsertReturning(ctx, "module_version_states", cols, vals, conflictAction,
 			[]string{"module_path", "version", "status"},
@@ -280,15 +310,20 @@ const moduleVersionStateColumns = `
 func scanModuleVersionState(scan func(dest ...interface{}) error) (*internal.ModuleVersionState, error) {
 	var (
 		v               internal.ModuleVersionState
+		indexTimestamp  pq.NullTime
 		lastProcessedAt pq.NullTime
 		numPackages     sql.NullInt64
 		hasGoMod        sql.NullBool
 	)
-	if err := scan(&v.ModulePath, &v.Version, &v.IndexTimestamp, &v.CreatedAt, &v.Status, &v.Error,
+	if err := scan(&v.ModulePath, &v.Version, &indexTimestamp, &v.CreatedAt, &v.Status, &v.Error,
 		&v.TryCount, &v.LastProcessedAt, &v.NextProcessedAfter, &v.AppVersion, &hasGoMod, &v.GoModPath,
 		&numPackages); err != nil {
 		return nil, err
 	}
+	if indexTimestamp.Valid {
+		it := indexTimestamp.Time
+		v.IndexTimestamp = &it
+	}
 	if lastProcessedAt.Valid {
 		lp := lastProcessedAt.Time
 		v.LastProcessedAt = &lp