insights: store: add support for recording data points

enterprise/internal/insights/resolvers/insight_series_resolver_test.go

@@ -110,6 +110,6 @@ func TestResolver_InsightSeries(t *testing.T) {
 		if err != nil {
 			t.Fatal(err)
 		}
-		autogold.Want("insights[0][0].Points mocked", "[{p:{Time:{wall:0 ext:63271811045 loc:<nil>} Value:1}} {p:{Time:{wall:0 ext:63271811045 loc:<nil>} Value:2}} {p:{Time:{wall:0 ext:63271811045 loc:<nil>} Value:3}}]").Equal(t, fmt.Sprintf("%+v", points))
+		autogold.Want("insights[0][0].Points mocked", "[{p:{Time:{wall:0 ext:63271811045 loc:<nil>} Value:1 Metadata:[]}} {p:{Time:{wall:0 ext:63271811045 loc:<nil>} Value:2 Metadata:[]}} {p:{Time:{wall:0 ext:63271811045 loc:<nil>} Value:3 Metadata:[]}}]").Equal(t, fmt.Sprintf("%+v", points))
 	})
 }

enterprise/internal/insights/store/store.go

@@ -3,11 +3,14 @@ package store
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"fmt"
 	"time"
 
 	"github.com/keegancsmith/sqlf"
+	"github.com/pkg/errors"
 
+	"github.com/sourcegraph/sourcegraph/internal/api"
 	"github.com/sourcegraph/sourcegraph/internal/database/basestore"
 	"github.com/sourcegraph/sourcegraph/internal/database/dbutil"
 	"github.com/sourcegraph/sourcegraph/internal/timeutil"
@@ -17,6 +20,7 @@ import (
 // for actual API usage.
 type Interface interface {
 	SeriesPoints(ctx context.Context, opts SeriesPointsOpts) ([]SeriesPoint, error)
+	RecordSeriesPoint(ctx context.Context, v RecordSeriesPointArgs) error
 }
 
 var _ Interface = &Store{}
@@ -55,15 +59,27 @@ func (s *Store) With(other basestore.ShareableStore) *Store {
 var _ Interface = &Store{}
 
 // SeriesPoint describes a single insights' series data point.
+//
+// Some fields that could be queried (series ID, repo ID/names) are omitted as they are primarily
+// only useful for filtering the data you get back, and would inflate the data size considerably
+// otherwise.
 type SeriesPoint struct {
-	Time  time.Time
-	Value float64
+	Time     time.Time
+	Value    float64
+	Metadata []byte
+}
+
+func (s *SeriesPoint) String() string {
+	return fmt.Sprintf("SeriesPoint{Time: %q, Value: %v, Metadata: %s}", s.Time, s.Value, s.Metadata)
 }
 
 // SeriesPointsOpts describes options for querying insights' series data points.
 type SeriesPointsOpts struct {
 	// SeriesID is the unique series ID to query, if non-nil.
-	SeriesID *int32
+	SeriesID *string
+
+	// TODO(slimsag): Add ability to filter based on repo ID, name, original name.
+	// TODO(slimsag): Add ability to do limited filtering based on metadata.
 
 	// Time ranges to query from/to, if non-nil.
 	From, To *time.Time
@@ -80,6 +96,7 @@ func (s *Store) SeriesPoints(ctx context.Context, opts SeriesPointsOpts) ([]Seri
 		err := sc.Scan(
 			&point.Time,
 			&point.Value,
+			&point.Metadata,
 		)
 		if err != nil {
 			return err
@@ -91,8 +108,12 @@ func (s *Store) SeriesPoints(ctx context.Context, opts SeriesPointsOpts) ([]Seri
 }
 
 var seriesPointsQueryFmtstr = `
--- source: enterprise/internal/insights/store/series_points.go
-SELECT time, value FROM series_points
+-- source: enterprise/internal/insights/store/store.go:SeriesPoints
+SELECT time,
+	value,
+	m.metadata
+FROM series_points p
+INNER JOIN metadata m ON p.metadata_id = m.id
 WHERE %s
 ORDER BY time DESC
 `
@@ -123,6 +144,127 @@ func seriesPointsQuery(opts SeriesPointsOpts) *sqlf.Query {
 	)
 }
 
+// RecordSeriesPointArgs describes arguments for the RecordSeriesPoint method.
+type RecordSeriesPointArgs struct {
+	// SeriesID is the unique series ID to query. It should describe the series of data uniquely,
+	// but is not a DB table primary key ID.
+	SeriesID string
+
+	// Point is the actual data point recorded and at what time.
+	Point SeriesPoint
+
+	// Repository name and DB ID to associate with this data point, if any.
+	//
+	// Both must be specified if one is specified.
+	RepoName *string
+	RepoID   *api.RepoID
+
+	// Metadata contains arbitrary JSON metadata to associate with the data point, if any.
+	//
+	// See the DB schema comments for intended use cases. This should generally be small,
+	// low-cardinality data to avoid inflating the table.
+	Metadata interface{}
+}
+
+// RecordSeriesPoint records a data point for the specfied series ID (which is a unique ID for the
+// series, not a DB table primary key ID).
+func (s *Store) RecordSeriesPoint(ctx context.Context, v RecordSeriesPointArgs) (err error) {
+	// Start transaction.
+	var txStore *basestore.Store
+	txStore, err = s.Transact(ctx)
+	if err != nil {
+		return err
+	}
+	defer func() { err = txStore.Done(err) }()
+
+	if (v.RepoName != nil && v.RepoID == nil) || (v.RepoID != nil && v.RepoName == nil) {
+		return errors.New("RepoName and RepoID must be mutually specified")
+	}
+
+	// Upsert the repository name into a separate table, so we get a small ID we can reference
+	// many times from the series_points table without storing the repo name multiple times.
+	var repoNameID *int
+	if v.RepoName != nil {
+		repoNameIDValue, ok, err := basestore.ScanFirstInt(txStore.Query(ctx, sqlf.Sprintf(upsertRepoNameFmtStr, *v.RepoName, *v.RepoName)))
+		if err != nil {
+			return errors.Wrap(err, "upserting repo name ID")
+		}
+		if !ok {
+			return errors.Wrap(err, "repo name ID not found (this should never happen)")
+		}
+		repoNameID = &repoNameIDValue
+	}
+
+	// Upsert the metadata into a separate table, so we get a small ID we can reference many times
+	// from the series_points table without storing the metadata multiple times.
+	var metadataID *int
+	if v.Metadata != nil {
+		jsonMetadata, err := json.Marshal(v.Metadata)
+		if err != nil {
+			return errors.Wrap(err, "upserting: encoding metadata")
+		}
+		metadataIDValue, ok, err := basestore.ScanFirstInt(txStore.Query(ctx, sqlf.Sprintf(upsertMetadataFmtStr, jsonMetadata, jsonMetadata)))
+		if err != nil {
+			return errors.Wrap(err, "upserting metadata ID")
+		}
+		if !ok {
+			return errors.Wrap(err, "metadata ID not found (this should never happen)")
+		}
+		metadataID = &metadataIDValue
+	}
+
+	// Insert the actual data point.
+	return txStore.Exec(ctx, sqlf.Sprintf(
+		recordSeriesPointFmtstr,
+		v.SeriesID,    // series_id
+		v.Point.Time,  // time
+		v.Point.Value, // value
+		metadataID,    // metadata_id
+		v.RepoID,      // repo_id
+		repoNameID,    // repo_name_id
+		repoNameID,    // original_repo_name_id
+	))
+}
+
+const upsertRepoNameFmtStr = `
+-- source: enterprise/internal/insights/store/store.go:RecordSeriesPoint
+WITH e AS(
+	INSERT INTO repo_names(name)
+	VALUES (%s)
+	ON CONFLICT DO NOTHING
+	RETURNING id
+)
+SELECT * FROM e
+UNION
+	SELECT id FROM repo_names WHERE name = %s;
+`
+
+const upsertMetadataFmtStr = `
+-- source: enterprise/internal/insights/store/store.go:RecordSeriesPoint
+WITH e AS(
+    INSERT INTO metadata(metadata)
+    VALUES (%s)
+    ON CONFLICT DO NOTHING
+    RETURNING id
+)
+SELECT * FROM e
+UNION
+	SELECT id FROM metadata WHERE metadata = %s;
+`
+
+const recordSeriesPointFmtstr = `
+-- source: enterprise/internal/insights/store/store.go:RecordSeriesPoint
+INSERT INTO series_points(
+	series_id,
+	time,
+	value,
+	metadata_id,
+	repo_id,
+	repo_name_id,
+	original_repo_name_id)
+VALUES (%s, %s, %s, %s, %s, %s, %s);
+`
+
 func (s *Store) query(ctx context.Context, q *sqlf.Query, sc scanFunc) error {
 	rows, err := s.Store.Query(ctx, q)
 	if err != nil {