store: scope activities to a user_id
garmin_activity_id uniqueness becomes per-user (UNIQUE(user_id, garmin_activity_id), added in Task 1) so two users' Garmin accounts can never collide even in the unlikely event their activity ids coincide.
This commit is contained in:
@@ -49,9 +49,9 @@ type Activity struct {
|
|||||||
UpdatedAt string
|
UpdatedAt string
|
||||||
}
|
}
|
||||||
|
|
||||||
// UpsertActivity inserts a new activity or updates the existing row for the
|
// UpsertActivity inserts a new activity for userID or updates the existing
|
||||||
// same garmin_activity_id (idempotent, safe to call on every sync pass), and
|
// row for the same (userID, garmin_activity_id) pair (idempotent, safe to
|
||||||
// returns its internal id.
|
// call on every sync pass), and returns its internal id.
|
||||||
func (db *DB) UpsertActivity(ctx context.Context, userID int64, a Activity) (int64, error) {
|
func (db *DB) UpsertActivity(ctx context.Context, userID int64, a Activity) (int64, error) {
|
||||||
_, err := db.ExecContext(ctx, `
|
_, err := db.ExecContext(ctx, `
|
||||||
INSERT INTO activities (
|
INSERT INTO activities (
|
||||||
@@ -61,7 +61,7 @@ func (db *DB) UpsertActivity(ctx context.Context, userID int64, a Activity) (int
|
|||||||
aerobic_training_effect, anaerobic_training_effect, vo2max_value,
|
aerobic_training_effect, anaerobic_training_effect, vo2max_value,
|
||||||
raw_json, updated_at
|
raw_json, updated_at
|
||||||
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?, datetime('now'))
|
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?, datetime('now'))
|
||||||
ON CONFLICT(garmin_activity_id) DO UPDATE SET
|
ON CONFLICT(user_id, garmin_activity_id) DO UPDATE SET
|
||||||
event_type_key=excluded.event_type_key,
|
event_type_key=excluded.event_type_key,
|
||||||
workout_id=excluded.workout_id,
|
workout_id=excluded.workout_id,
|
||||||
start_time_utc=excluded.start_time_utc,
|
start_time_utc=excluded.start_time_utc,
|
||||||
@@ -88,8 +88,8 @@ func (db *DB) UpsertActivity(ctx context.Context, userID int64, a Activity) (int
|
|||||||
}
|
}
|
||||||
|
|
||||||
var id int64
|
var id int64
|
||||||
if err := db.QueryRowContext(ctx, `SELECT id FROM activities WHERE garmin_activity_id = ? AND user_id = ?`, a.GarminActivityID, userID).Scan(&id); err != nil {
|
if err := db.QueryRowContext(ctx, `SELECT id FROM activities WHERE user_id = ? AND garmin_activity_id = ?`, userID, a.GarminActivityID).Scan(&id); err != nil {
|
||||||
return 0, fmt.Errorf("fetch id for activity %d: %w", a.GarminActivityID, err)
|
return 0, fmt.Errorf("fetch id for activity %d (user %d): %w", a.GarminActivityID, userID, err)
|
||||||
}
|
}
|
||||||
return id, nil
|
return id, nil
|
||||||
}
|
}
|
||||||
@@ -116,15 +116,15 @@ const activityColumns = `
|
|||||||
created_at, updated_at
|
created_at, updated_at
|
||||||
`
|
`
|
||||||
|
|
||||||
// GetActivity fetches one activity by its internal id.
|
// GetActivity fetches one activity by its internal id, scoped to userID.
|
||||||
func (db *DB) GetActivity(ctx context.Context, id int64) (Activity, bool, error) {
|
func (db *DB) GetActivity(ctx context.Context, userID, id int64) (Activity, bool, error) {
|
||||||
row := db.QueryRowContext(ctx, `SELECT `+activityColumns+` FROM activities WHERE id = ?`, id)
|
row := db.QueryRowContext(ctx, `SELECT `+activityColumns+` FROM activities WHERE id = ? AND user_id = ?`, id, userID)
|
||||||
a, err := scanActivity(row)
|
a, err := scanActivity(row)
|
||||||
if err == sql.ErrNoRows {
|
if err == sql.ErrNoRows {
|
||||||
return Activity{}, false, nil
|
return Activity{}, false, nil
|
||||||
}
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return Activity{}, false, fmt.Errorf("get activity %d: %w", id, err)
|
return Activity{}, false, fmt.Errorf("get activity %d for user %d: %w", id, userID, err)
|
||||||
}
|
}
|
||||||
return a, true, nil
|
return a, true, nil
|
||||||
}
|
}
|
||||||
@@ -137,7 +137,8 @@ type ActivityFilter struct {
|
|||||||
Offset int
|
Offset int
|
||||||
}
|
}
|
||||||
|
|
||||||
// ListActivities returns activities for userID newest-first, optionally filtered by date range.
|
// ListActivities returns userID's activities newest-first, optionally
|
||||||
|
// filtered by date range.
|
||||||
func (db *DB) ListActivities(ctx context.Context, userID int64, f ActivityFilter) ([]Activity, error) {
|
func (db *DB) ListActivities(ctx context.Context, userID int64, f ActivityFilter) ([]Activity, error) {
|
||||||
query := `SELECT ` + activityColumns + ` FROM activities WHERE user_id = ?`
|
query := `SELECT ` + activityColumns + ` FROM activities WHERE user_id = ?`
|
||||||
args := []any{userID}
|
args := []any{userID}
|
||||||
@@ -172,83 +173,79 @@ func (db *DB) ListActivities(ctx context.Context, userID int64, f ActivityFilter
|
|||||||
return activities, rows.Err()
|
return activities, rows.Err()
|
||||||
}
|
}
|
||||||
|
|
||||||
// ActivityExists reports whether an activity with this garmin_activity_id is
|
// ActivityExists reports whether userID already has an activity with this
|
||||||
// already stored, checked before each upsert during a sync pass so the
|
// garmin_activity_id stored.
|
||||||
// reported "activities fetched" count reflects genuinely new activities, not
|
func (db *DB) ActivityExists(ctx context.Context, userID, garminActivityID int64) (bool, error) {
|
||||||
// every activity Garmin's API happens to return for the queried date range
|
|
||||||
// (which, thanks to the incremental overlap window and backfill's
|
|
||||||
// already-covered history, is almost always a re-listing of known ones).
|
|
||||||
func (db *DB) ActivityExists(ctx context.Context, garminActivityID int64) (bool, error) {
|
|
||||||
var id int64
|
var id int64
|
||||||
err := db.QueryRowContext(ctx, `SELECT id FROM activities WHERE garmin_activity_id = ?`, garminActivityID).Scan(&id)
|
err := db.QueryRowContext(ctx, `SELECT id FROM activities WHERE user_id = ? AND garmin_activity_id = ?`, userID, garminActivityID).Scan(&id)
|
||||||
if err == sql.ErrNoRows {
|
if err == sql.ErrNoRows {
|
||||||
return false, nil
|
return false, nil
|
||||||
}
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, fmt.Errorf("check activity %d exists: %w", garminActivityID, err)
|
return false, fmt.Errorf("check activity %d exists for user %d: %w", garminActivityID, userID, err)
|
||||||
}
|
}
|
||||||
return true, nil
|
return true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// LatestActivityStartTime returns the start_time_utc of the most recently
|
// LatestActivityStartTime returns userID's most recently started activity's
|
||||||
// started activity we have, used to compute the incremental sync window.
|
// start_time_utc, used to compute the incremental sync window.
|
||||||
func (db *DB) LatestActivityStartTime(ctx context.Context) (string, bool, error) {
|
func (db *DB) LatestActivityStartTime(ctx context.Context, userID int64) (string, bool, error) {
|
||||||
var t string
|
var t string
|
||||||
err := db.QueryRowContext(ctx, `SELECT start_time_utc FROM activities ORDER BY start_time_utc DESC LIMIT 1`).Scan(&t)
|
err := db.QueryRowContext(ctx, `SELECT start_time_utc FROM activities WHERE user_id = ? ORDER BY start_time_utc DESC LIMIT 1`, userID).Scan(&t)
|
||||||
if err == sql.ErrNoRows {
|
if err == sql.ErrNoRows {
|
||||||
return "", false, nil
|
return "", false, nil
|
||||||
}
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return "", false, fmt.Errorf("latest activity start time: %w", err)
|
return "", false, fmt.Errorf("latest activity start time for user %d: %w", userID, err)
|
||||||
}
|
}
|
||||||
return t, true, nil
|
return t, true, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetActivityDetails records that get_activity_details has been fetched for
|
// SetActivityDetails records that get_activity_details has been fetched for
|
||||||
// this activity, storing the raw response for future reprocessing.
|
// this activity, storing the raw response for future reprocessing.
|
||||||
func (db *DB) SetActivityDetails(ctx context.Context, activityID int64, rawJSON string) error {
|
func (db *DB) SetActivityDetails(ctx context.Context, userID, activityID int64, rawJSON string) error {
|
||||||
_, err := db.ExecContext(ctx, `
|
_, err := db.ExecContext(ctx, `
|
||||||
UPDATE activities SET details_fetched_at = datetime('now'), details_raw_json = ?, updated_at = datetime('now')
|
UPDATE activities SET details_fetched_at = datetime('now'), details_raw_json = ?, updated_at = datetime('now')
|
||||||
WHERE id = ?`, rawJSON, activityID)
|
WHERE id = ? AND user_id = ?`, rawJSON, activityID, userID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("set activity %d details: %w", activityID, err)
|
return fmt.Errorf("set activity %d details for user %d: %w", activityID, userID, err)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetActivityWorkout stores the raw get_workout_by_id() response used to
|
// SetActivityWorkout stores the raw get_workout_by_id() response used to
|
||||||
// compute this activity's laps' target pace/HR bands.
|
// compute this activity's laps' target pace/HR bands.
|
||||||
func (db *DB) SetActivityWorkout(ctx context.Context, activityID int64, rawJSON string) error {
|
func (db *DB) SetActivityWorkout(ctx context.Context, userID, activityID int64, rawJSON string) error {
|
||||||
_, err := db.ExecContext(ctx, `
|
_, err := db.ExecContext(ctx, `
|
||||||
UPDATE activities SET workout_raw_json = ?, updated_at = datetime('now')
|
UPDATE activities SET workout_raw_json = ?, updated_at = datetime('now')
|
||||||
WHERE id = ?`, rawJSON, activityID)
|
WHERE id = ? AND user_id = ?`, rawJSON, activityID, userID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("set activity %d workout: %w", activityID, err)
|
return fmt.Errorf("set activity %d workout for user %d: %w", activityID, userID, err)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetActivitySplitsFetched records that get_activity_splits has been fetched
|
// SetActivitySplitsFetched records that get_activity_splits has been fetched
|
||||||
// for this activity.
|
// for this activity.
|
||||||
func (db *DB) SetActivitySplitsFetched(ctx context.Context, activityID int64) error {
|
func (db *DB) SetActivitySplitsFetched(ctx context.Context, userID, activityID int64) error {
|
||||||
_, err := db.ExecContext(ctx, `
|
_, err := db.ExecContext(ctx, `
|
||||||
UPDATE activities SET splits_fetched_at = datetime('now'), updated_at = datetime('now')
|
UPDATE activities SET splits_fetched_at = datetime('now'), updated_at = datetime('now')
|
||||||
WHERE id = ?`, activityID)
|
WHERE id = ? AND user_id = ?`, activityID, userID)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("set activity %d splits fetched: %w", activityID, err)
|
return fmt.Errorf("set activity %d splits fetched for user %d: %w", activityID, userID, err)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// ActivitiesMissingDetails returns activities that haven't had
|
// ActivitiesMissingDetails returns userID's activities that haven't had
|
||||||
// get_activity_details/get_activity_splits fetched yet, for the lazy
|
// get_activity_details/get_activity_splits fetched yet, for the lazy
|
||||||
// background detail-fill pass.
|
// background detail-fill pass.
|
||||||
func (db *DB) ActivitiesMissingDetails(ctx context.Context, limit int) ([]Activity, error) {
|
func (db *DB) ActivitiesMissingDetails(ctx context.Context, userID int64, limit int) ([]Activity, error) {
|
||||||
rows, err := db.QueryContext(ctx, `SELECT `+activityColumns+` FROM activities
|
rows, err := db.QueryContext(ctx, `SELECT `+activityColumns+` FROM activities
|
||||||
WHERE details_fetched_at IS NULL OR splits_fetched_at IS NULL
|
WHERE user_id = ? AND (details_fetched_at IS NULL OR splits_fetched_at IS NULL)
|
||||||
ORDER BY start_time_utc DESC LIMIT ?`, limit)
|
ORDER BY start_time_utc DESC LIMIT ?`, userID, limit)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("list activities missing details: %w", err)
|
return nil, fmt.Errorf("list activities missing details for user %d: %w", userID, err)
|
||||||
}
|
}
|
||||||
defer rows.Close()
|
defer rows.Close()
|
||||||
|
|
||||||
@@ -263,15 +260,15 @@ func (db *DB) ActivitiesMissingDetails(ctx context.Context, limit int) ([]Activi
|
|||||||
return activities, rows.Err()
|
return activities, rows.Err()
|
||||||
}
|
}
|
||||||
|
|
||||||
// CountActivitiesMissingDetails returns how many activities still need
|
// CountActivitiesMissingDetails returns how many of userID's activities
|
||||||
// get_activity_details/get_activity_splits fetched, regardless of any
|
// still need get_activity_details/get_activity_splits fetched, regardless
|
||||||
// per-call batch limit -- used to report overall remaining work.
|
// of any per-call batch limit -- used to report overall remaining work.
|
||||||
func (db *DB) CountActivitiesMissingDetails(ctx context.Context) (int, error) {
|
func (db *DB) CountActivitiesMissingDetails(ctx context.Context, userID int64) (int, error) {
|
||||||
var n int
|
var n int
|
||||||
err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM activities
|
err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM activities
|
||||||
WHERE details_fetched_at IS NULL OR splits_fetched_at IS NULL`).Scan(&n)
|
WHERE user_id = ? AND (details_fetched_at IS NULL OR splits_fetched_at IS NULL)`, userID).Scan(&n)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return 0, fmt.Errorf("count activities missing details: %w", err)
|
return 0, fmt.Errorf("count activities missing details for user %d: %w", userID, err)
|
||||||
}
|
}
|
||||||
return n, nil
|
return n, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -41,7 +41,7 @@ func TestMigrateIsIdempotent(t *testing.T) {
|
|||||||
func TestUpsertActivity_InsertThenUpdateIsIdempotent(t *testing.T) {
|
func TestUpsertActivity_InsertThenUpdateIsIdempotent(t *testing.T) {
|
||||||
db := openTestDB(t)
|
db := openTestDB(t)
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
userID, err := db.ProvisionUser(ctx, "test-sub", "Test User")
|
userID, err := db.ProvisionUser(ctx, "test-sub", "Test")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("ProvisionUser: %v", err)
|
t.Fatalf("ProvisionUser: %v", err)
|
||||||
}
|
}
|
||||||
@@ -69,7 +69,7 @@ func TestUpsertActivity_InsertThenUpdateIsIdempotent(t *testing.T) {
|
|||||||
t.Fatalf("expected same internal id across upserts, got %d then %d", id, id2)
|
t.Fatalf("expected same internal id across upserts, got %d then %d", id, id2)
|
||||||
}
|
}
|
||||||
|
|
||||||
got, ok, err := db.GetActivity(ctx, id)
|
got, ok, err := db.GetActivity(ctx, userID, id)
|
||||||
if err != nil || !ok {
|
if err != nil || !ok {
|
||||||
t.Fatalf("GetActivity: ok=%v err=%v", ok, err)
|
t.Fatalf("GetActivity: ok=%v err=%v", ok, err)
|
||||||
}
|
}
|
||||||
@@ -86,10 +86,40 @@ func TestUpsertActivity_InsertThenUpdateIsIdempotent(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestUpsertActivity_SameGarminActivityIDAllowedAcrossDifferentUsers(t *testing.T) {
|
||||||
|
db := openTestDB(t)
|
||||||
|
ctx := context.Background()
|
||||||
|
userA, err := db.ProvisionUser(ctx, "sub-a", "A")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ProvisionUser(a): %v", err)
|
||||||
|
}
|
||||||
|
userB, err := db.ProvisionUser(ctx, "sub-b", "B")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("ProvisionUser(b): %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
activity := Activity{GarminActivityID: 999, StartTimeUTC: "2026-07-11 05:00:00", RawJSON: "{}"}
|
||||||
|
if _, err := db.UpsertActivity(ctx, userA, activity); err != nil {
|
||||||
|
t.Fatalf("UpsertActivity(a): %v", err)
|
||||||
|
}
|
||||||
|
if _, err := db.UpsertActivity(ctx, userB, activity); err != nil {
|
||||||
|
t.Fatalf("expected the same garmin_activity_id to be allowed for a different user, got: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
aActivities, err := db.ListActivities(ctx, userA, ActivityFilter{})
|
||||||
|
if err != nil || len(aActivities) != 1 {
|
||||||
|
t.Fatalf("ListActivities(a): got %d, err=%v", len(aActivities), err)
|
||||||
|
}
|
||||||
|
bActivities, err := db.ListActivities(ctx, userB, ActivityFilter{})
|
||||||
|
if err != nil || len(bActivities) != 1 {
|
||||||
|
t.Fatalf("ListActivities(b): got %d, err=%v", len(bActivities), err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestKindAssignment_AppendOnlyHistoryAndCurrentView(t *testing.T) {
|
func TestKindAssignment_AppendOnlyHistoryAndCurrentView(t *testing.T) {
|
||||||
db := openTestDB(t)
|
db := openTestDB(t)
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
userID, err := db.ProvisionUser(ctx, "test-sub", "Test User")
|
userID, err := db.ProvisionUser(ctx, "test-sub", "Test")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("ProvisionUser: %v", err)
|
t.Fatalf("ProvisionUser: %v", err)
|
||||||
}
|
}
|
||||||
@@ -177,7 +207,7 @@ func TestKindAssignment_AppendOnlyHistoryAndCurrentView(t *testing.T) {
|
|||||||
func TestReplaceLapsIsIdempotent(t *testing.T) {
|
func TestReplaceLapsIsIdempotent(t *testing.T) {
|
||||||
db := openTestDB(t)
|
db := openTestDB(t)
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
userID, err := db.ProvisionUser(ctx, "test-sub", "Test User")
|
userID, err := db.ProvisionUser(ctx, "test-sub", "Test")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("ProvisionUser: %v", err)
|
t.Fatalf("ProvisionUser: %v", err)
|
||||||
}
|
}
|
||||||
@@ -257,7 +287,7 @@ func TestForeignKeyEnforcementPostMigration(t *testing.T) {
|
|||||||
// migrations 0023/0024/0025) is properly scoped and re-enabled after each
|
// migrations 0023/0024/0025) is properly scoped and re-enabled after each
|
||||||
// table rebuild.
|
// table rebuild.
|
||||||
|
|
||||||
userID, err := db.ProvisionUser(ctx, "test-sub", "Test User")
|
userID, err := db.ProvisionUser(ctx, "test-sub", "Test")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("ProvisionUser: %v", err)
|
t.Fatalf("ProvisionUser: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user