vercel · jaredpalmer · Feb 25, 2022 · Feb 17, 2022 · Feb 17, 2022 · Feb 17, 2022
@@ -6,11 +6,10 @@ on:
   pull_request:
     types: [opened, synchronize]
 
-
 jobs:
   build:
     name: build and test
-    timeout-minutes: 10
+    timeout-minutes: 15
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:

@@ -14,7 +14,7 @@ require (
 	github.com/gobwas/glob v0.2.3
 	github.com/google/chrometracing v0.0.0-20210413150014-55fded0163e7
 	github.com/google/go-cmp v0.5.5 // indirect
-	github.com/google/uuid v1.2.0 // indirect
+	github.com/google/uuid v1.3.0 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-gatedio v0.5.0
 	github.com/hashicorp/go-hclog v1.1.0
@@ -25,6 +25,7 @@ require (
 	github.com/mattn/go-isatty v0.0.12
 	github.com/mitchellh/cli v1.1.2
 	github.com/mitchellh/go-homedir v1.1.0
+	github.com/mitchellh/mapstructure v1.4.3 // indirect
 	github.com/mitchellh/reflectwalk v1.0.1 // indirect
 	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
 	github.com/pkg/errors v0.9.1

@@ -38,6 +38,8 @@ github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
 github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
+github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
 github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=
 github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
@@ -90,6 +92,8 @@ github.com/mitchellh/copystructure v1.0.0 h1:Laisrj+bAB6b/yJwB5Bt3ITZhGJdqmxquMK
 github.com/mitchellh/copystructure v1.0.0/go.mod h1:SNtv71yrdKgLRyLFxmLdkAbkKEFWgYaq1OVrnRcwhnw=
 github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
+github.com/mitchellh/mapstructure v1.4.3 h1:OVowDSCllw/YjdLkam3/sm7wEtOy59d8ndGgCcyj8cs=
+github.com/mitchellh/mapstructure v1.4.3/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
 github.com/mitchellh/reflectwalk v1.0.0/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=
 github.com/mitchellh/reflectwalk v1.0.1 h1:FVzMWA5RllMAKIdUSC8mdWo3XtwoecrH79BY70sEEpE=
 github.com/mitchellh/reflectwalk v1.0.1/go.mod h1:mSTlrgnPZtwu0c4WaC2kGObEpuNDbx0jmZXqmk4esnw=

@@ -0,0 +1,175 @@
+package analytics
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/hashicorp/go-hclog"
+	"github.com/mitchellh/mapstructure"
+	"github.com/vercel/turborepo/cli/internal/util"
+)
+
+type Events = []map[string]interface{}
+
+type EventPayload = interface{}
+
+type Recorder interface {
+	LogEvent(payload EventPayload)
+}
+
+type Client interface {
+	Recorder
+	Close()
+	CloseWithTimeout(timeout time.Duration)
+}
+
+type Sink interface {
+	RecordAnalyticsEvents(events Events) error
+}
+
+type nullSink struct{}
+
+func (n *nullSink) RecordAnalyticsEvents(events Events) error {
+	return nil
+}
+
+// NullSink is an analytics sink to use in the event that we don't want to send
+// analytics
+var NullSink = &nullSink{}
+
+type client struct {
+	ch     chan<- EventPayload
+	cancel func()
+
+	worker *worker
+}
+
+type worker struct {
+	buffer        []EventPayload
+	ch            <-chan EventPayload
+	ctx           context.Context
+	doneSemaphore util.Semaphore
+	sessionID     uuid.UUID
+	sink          Sink
+	wg            sync.WaitGroup
+	logger        hclog.Logger
+}
+
+const bufferThreshold = 10
+const eventTimeout = 200 * time.Millisecond
+const noTimeout = 24 * time.Hour
+
+func newWorker(ctx context.Context, ch <-chan EventPayload, sink Sink, logger hclog.Logger) *worker {
+	buffer := []EventPayload{}
+	sessionID := uuid.New()
+	w := &worker{
+		buffer:        buffer,
+		ch:            ch,
+		ctx:           ctx,
+		doneSemaphore: util.NewSemaphore(1),
+		sessionID:     sessionID,
+		sink:          sink,
+		logger:        logger,
+	}
+	w.doneSemaphore.Acquire()
+	go w.analyticsClient()
+	return w
+}
+
+func NewClient(parent context.Context, sink Sink, logger hclog.Logger) Client {
+	ch := make(chan EventPayload)
+	ctx, cancel := context.WithCancel(parent)
+	// creates and starts the worker
+	worker := newWorker(ctx, ch, sink, logger)
+	s := &client{
+		ch:     ch,
+		cancel: cancel,
+		worker: worker,
+	}
+	return s
+}
+
+func (s *client) LogEvent(event EventPayload) {
+	s.ch <- event
+}
+
+func (s *client) Close() {
+	s.cancel()
+	s.worker.Wait()
+}
+
+func (s *client) CloseWithTimeout(timeout time.Duration) {
+	ch := make(chan struct{})
+	go func() {
+		s.Close()
+		close(ch)
+	}()
+	select {
+	case <-ch:
+	case <-time.After(timeout):
+	}
+}
+
+func (w *worker) Wait() {
+	w.doneSemaphore.Acquire()
+	w.wg.Wait()
+}
+
+func (w *worker) analyticsClient() {
+	timeout := time.After(noTimeout)
+	for {
+		select {
+		case e := <-w.ch:
+			w.buffer = append(w.buffer, e)
+			if len(w.buffer) == bufferThreshold {
+				w.flush()
+				timeout = time.After(noTimeout)
+			} else {
+				timeout = time.After(eventTimeout)
+			}
+		case <-timeout:
+			w.flush()
+			timeout = time.After(noTimeout)
+		case <-w.ctx.Done():
+			w.flush()
+			w.doneSemaphore.Release()
+			return
+		}
+	}
+}
+
+func (w *worker) flush() {
+	if len(w.buffer) > 0 {
+		w.sendEvents(w.buffer)
+		w.buffer = []EventPayload{}
+	}
+}
+
+func (w *worker) sendEvents(events []EventPayload) {
+	w.wg.Add(1)
+	go func() {
+		payload, err := addSessionID(w.sessionID.String(), events)
+		if err != nil {
+			w.logger.Debug("failed to encode cache usage analytics", "error", err)
+		}
+		err = w.sink.RecordAnalyticsEvents(payload)
+		if err != nil {
+			w.logger.Debug("failed to record cache usage analytics", "error", err)
+		}
+		w.wg.Done()
+	}()
+}
+
+func addSessionID(sessionID string, events []EventPayload) (Events, error) {
+	eventMaps := []map[string]interface{}{}
+	err := mapstructure.Decode(events, &eventMaps)
+	if err != nil {
+		return nil, err
+	}
+	for _, event := range eventMaps {
+		event["sessionId"] = sessionID
+	}
+	return eventMaps, nil
+}