Assign incoming requests a session ID

Signed-off-by: AKP <tom@tdpain.net>

analytics/db/migrations/20230331164907_initial.go

@@ -12,6 +12,7 @@ func init() {
 
 	tables := []any{
 		&models.Request{},
+		&models.Session{},
 	}
 
 	Migrations.MustRegister(func(ctx context.Context, db *bun.DB) error {

analytics/db/models/request.go

@@ -16,4 +16,7 @@ type Request struct {
 	URI       string
 	Referer   string
 	UserAgent string
+
+	Session   *Session `bun:"rel:belongs-to,join:session_id=id"`
+	SessionID string   `bun:",nullzero"`
 }

analytics/db/models/session.go

@@ -0,0 +1,13 @@
+package models
+
+import (
+	"github.com/uptrace/bun"
+)
+
+type Session struct {
+	bun.BaseModel
+
+	ID        string `bun:",pk"`
+	UserAgent string `bun:"type:VARCHAR COLLATE NOCASE"`
+	IPAddr    string
+}

analytics/ingest/log.go

@@ -2,6 +2,7 @@ package ingest
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
 	"github.com/codemicro/analytics/analytics/db/models"
 	"github.com/lithammer/shortuuid/v4"
@@ -23,10 +24,30 @@ func (i *Ingest) processLog(inp []byte) {
 	req, err := cl.ToRequestModel()
 	if err != nil {
 		log.Error().Err(err).Bytes("raw_json", inp).Msg("could not convert CaddyLog to Request")
+		return
 	}
 
-	if _, err := i.db.DB.NewInsert().Model(req).Exec(context.Background()); err != nil {
+	tx, err := i.db.DB.BeginTx(context.Background(), &sql.TxOptions{})
+	if err != nil {
+		log.Error().Err(err).Msg("failed to start transaction")
+		return
+	}
+
+	sess, err := i.assignToSession(tx, req)
+	if err != nil {
+		log.Error().Err(err).Msg("failed to assign session to request")
+	}
+
+	req.SessionID = sess.ID
+
+	if _, err := tx.NewInsert().Model(req).Exec(context.Background()); err != nil {
 		log.Error().Err(err).Msg("could not save request into database")
+		return
+	}
+
+	if err := tx.Commit(); err != nil {
+		log.Error().Err(err).Msg("unable to commit transaction")
+		return
 	}
 }
 

analytics/ingest/session.go

@@ -0,0 +1,40 @@
+package ingest
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"github.com/codemicro/analytics/analytics/db/models"
+	"github.com/lithammer/shortuuid/v4"
+	"github.com/uptrace/bun"
+)
+
+func (i *Ingest) assignToSession(tx bun.Tx, request *models.Request) (*models.Session, error) {
+	sess := new(models.Session)
+
+	err := tx.NewSelect().
+		Model(sess).
+		Where("ip_addr = ?", request.IPAddr).
+		Where("user_agent = ?", request.UserAgent).
+		Where(`? < datetime((select max("time") as "time" from requests where session_id = "session"."id"), '+30 minutes')`, request.Time).
+		Scan(context.Background(), sess)
+	if err != nil {
+		if !errors.Is(err, sql.ErrNoRows) {
+			return nil, err
+		}
+	} else {
+		return sess, nil
+	}
+
+	// No session found that matches, create a new one
+	sess.ID = shortuuid.New()
+	sess.IPAddr = request.IPAddr
+	sess.UserAgent = request.UserAgent
+
+	_, err = tx.NewInsert().Model(sess).Exec(context.Background())
+	if err != nil {
+		return nil, err
+	}
+
+	return sess, nil
+}