<commit_before>// This package provides the interface and functions to deal with the storage of data.
package store

import (
	"appengine"
	"appengine/datastore"
	"github.com/alexandre-normand/glukit/app/apimodel"
	"github.com/alexandre-normand/glukit/app/model"
	"github.com/alexandre-normand/glukit/app/util"
	"math"
	"time"
)

const (
	// Number of GlukitScores to batch in a single PutMulti
	GLUKIT_SCORE_PUT_MULTI_SIZE = 25
)

// Error interface to distinguish between temporary errors from permanent ones
type StoreError struct {
	msg       string
	Temporary bool // Is the error temporary?
}

func (e StoreError) Error() string {
	return e.msg
}

type GlukitScoreScanQuery struct {
	Limit *int
	From  *time.Time
	To    *time.Time
}

var (
	// ErrNoImportedDataFound is returned when the user doesn't have data imported yet.
	ErrNoImportedDataFound = StoreError{"store: no imported data found", true}

	// ErrNoSteadySailorMatchFound is returned when the user doesn't have any steady sailor matching his profile
	ErrNoSteadySailorMatchFound = StoreError{"store: no match for a steady sailor found", true}
)

// GetUserKey returns the GlukitUser datastore key given its email address.
func GetUserKey(context appengine.Context, email string) (key *datastore.Key) {
	return datastore.NewKey(context, "GlukitUser", email, 0, nil)
}

// StoreUserProfile stores a GlukitUser profile to the datastore. If the entry already exists, it is overriden and it is created
// otherwise
func StoreUserProfile(context appengine.Context, updatedAt time.Time, userProfile model.GlukitUser) (key *datastore.Key, err error) {
	key, error := datastore.Put(context, GetUserKey(context, userProfile.Email), &userProfile)
	if error != nil {
		util.Propagate(error)
	}

	return key, nil
}

// GetUserProfile returns the GlukitUser entry associated with the given datastore key. This can be obtained
// by calling GetUserKey.
func GetUserProfile(context appengine.Context, key *datastore.Key) (userProfile *model.GlukitUser, err error) {
	userProfile = new(model.GlukitUser)
	context.Infof("Fetching user profile for key: %s", key.String())
	error := datastore.Get(context, key, userProfile)
	if error != nil {
		return nil, error
	}

	return userProfile, nil
}

// StoreDaysOfReads stores a batch of DayOfReads elements. It is a optimized operation in that:
//    1. One element represents a relatively short-and-wide entry of all reads for a single day.
//    2. We have multiple DayOfReads elements and we use a PutMulti to make this faster.
// For details of how a single element of DayOfReads is physically stored, see the implementation of apimodel.Store and apimodel.Load.
// Also important to note, this store operation also handles updating the GlukitUser entry with the most recent read, if applicable.
func StoreDaysOfReads(context appengine.Context, userProfileKey *datastore.Key, daysOfReads []apimodel.DayOfGlucoseReads) (keys []*datastore.Key, err error) {
	elementKeys := make([]*datastore.Key, len(daysOfReads))
	for i := range daysOfReads {
		elementKeys[i] = datastore.NewKey(context, "DayOfReads", "", daysOfReads[i].Reads[0].Time.Timestamp, userProfileKey)
	}

	context.Infof("Emitting a PutMulti with %d keys for all %d days of reads", len(elementKeys), len(daysOfReads))
	keys, error := datastore.PutMulti(context, elementKeys, daysOfReads)
	if error != nil {
		context.Warningf("Error writing %d days of reads with keys [%s]: %v", len(elementKeys), elementKeys, error)
		return nil, error
	}

	// Get the time of the batch's last read and update the most recent read timestamp if necessary
	userProfile, err := GetGlukitUserWithKey(context, userProfileKey)
	if err != nil {
		context.Criticalf("Error reading user profile [%s] for its most recent read value: %v", userProfileKey, err)
		return nil, err
	}

	lastDayOfRead := daysOfReads[len(daysOfReads)-1]
	lastRead := lastDayOfRead.Reads[len(lastDayOfRead.Reads)-1]
	if userProfile.MostRecentRead.GetTime().Before(lastRead.GetTime()) {
		context.Infof("Updating most recent read date to %s", lastRead.GetTime())
		userProfile.MostRecentRead = lastRead
		_, err := StoreUserProfile(context, time.Now(), *userProfile)
		if err != nil {
			context.Criticalf("Error storing updated user profile [%s] with most recent read value of %s: %v", userProfileKey, userProfile.MostRecentRead, err)
			return nil, err
		}
	}

	return elementKeys, nil
}

// StoreCalibrationReads stores a batch of DayOfCalibrations elements. It is a optimized operation in that:
//    1. One element represents a relatively short-and-wide entry of all calibration reads for a single day.
//    2. We have multiple DayOfReads elements and we use a PutMulti to make this faster.
// For details of how a single element of DayOfReads is physically stored, see the implementation of apimodel.Store and apimodel.Load.
func StoreCalibrationReads(context appengine.Context, userProfileKey *datastore.Key, daysOfCalibrationReads []apimodel.DayOfCalibrationReads) (keys []*datastore.Key, err error) {
	elementKeys := make([]*datastore.Key, len(daysOfCalibrationReads))
	for i := range daysOfCalibrationReads {
		elementKeys[i] = datastore.NewKey(context, "DayOfCalibrationReads", "", daysOfCalibrationReads[i].Reads[0].Time.Timestamp, userProfileKey)
	}

	context.Infof("Emitting a PutMulti with %d keys for all %d days of calibration reads", len(elementKeys), len(daysOfCalibrationReads))
	keys, error := datastore.PutMulti(context, elementKeys, daysOfCalibrationReads)
	if error != nil {
		context.Criticalf("Error writing %d days of calibration reads with keys [%s]: %v", len(elementKeys), elementKeys, error)
		return nil, error
	}

	return elementKeys, nil
}

// GetGlucoseReads returns all GlucoseReads given a user's email address and the time boundaries. Not that the boundaries are both inclusive.
func GetGlucoseReads(context appengine.Context, email string, lowerBound time.Time, upperBound time.Time) (reads []apimodel.GlucoseRead, err error) {
	key := GetUserKey(context, email)

	// Scan start should be one day prior and scan end should be one day later so that we can capture the day using
	// a single column inequality filter. The scan should actually capture at least one day and a maximum of 3
	scanStart := lowerBound.Add(time.Duration(-24 * time.Hour))
	scanEnd := upperBound.Add(time.Duration(24 * time.Hour))

	context.Infof("Scanning for reads between %s and %s to get reads between %s and %s", scanStart, scanEnd, lowerBound, upperBound)

	query := datastore.NewQuery("DayOfReads").Ancestor(key).Filter("startTime >=", scanStart).Filter("startTime <=", scanEnd).Order("startTime")
	var daysOfReads apimodel.DayOfGlucoseReads
	readsForPeriod := make([]apimodel.GlucoseRead, 0)

	iterator := query.Run(context)
	for _, err := iterator.Next(&daysOfReads); err == nil; _, err = iterator.Next(&daysOfReads) {
		context.Debugf("Loaded batch of %d reads...", len(daysOfReads.Reads))
		readsForPeriod = mergeGlucoseReadArrays(readsForPeriod, daysOfReads.Reads)
	}<commit_msg>Fix Loading of GlucoseReads Leaking Data Across Batches.<commit_after>// This package provides the interface and functions to deal with the storage of data.
package store

import (
	"appengine"
	"appengine/datastore"
	"github.com/alexandre-normand/glukit/app/apimodel"
	"github.com/alexandre-normand/glukit/app/model"
	"github.com/alexandre-normand/glukit/app/util"
	"math"
	"time"
)

const (
	// Number of GlukitScores to batch in a single PutMulti
	GLUKIT_SCORE_PUT_MULTI_SIZE = 25
)

// Error interface to distinguish between temporary errors from permanent ones
type StoreError struct {
	msg       string
	Temporary bool // Is the error temporary?
}

func (e StoreError) Error() string {
	return e.msg
}

type GlukitScoreScanQuery struct {
	Limit *int
	From  *time.Time
	To    *time.Time
}

var (
	// ErrNoImportedDataFound is returned when the user doesn't have data imported yet.
	ErrNoImportedDataFound = StoreError{"store: no imported data found", true}

	// ErrNoSteadySailorMatchFound is returned when the user doesn't have any steady sailor matching his profile
	ErrNoSteadySailorMatchFound = StoreError{"store: no match for a steady sailor found", true}
)

// GetUserKey returns the GlukitUser datastore key given its email address.
func GetUserKey(context appengine.Context, email string) (key *datastore.Key) {
	return datastore.NewKey(context, "GlukitUser", email, 0, nil)
}

// StoreUserProfile stores a GlukitUser profile to the datastore. If the entry already exists, it is overriden and it is created
// otherwise
func StoreUserProfile(context appengine.Context, updatedAt time.Time, userProfile model.GlukitUser) (key *datastore.Key, err error) {
	key, error := datastore.Put(context, GetUserKey(context, userProfile.Email), &userProfile)
	if error != nil {
		util.Propagate(error)
	}

	return key, nil
}

// GetUserProfile returns the GlukitUser entry associated with the given datastore key. This can be obtained
// by calling GetUserKey.
func GetUserProfile(context appengine.Context, key *datastore.Key) (userProfile *model.GlukitUser, err error) {
	userProfile = new(model.GlukitUser)
	context.Infof("Fetching user profile for key: %s", key.String())
	error := datastore.Get(context, key, userProfile)
	if error != nil {
		return nil, error
	}

	return userProfile, nil
}

// StoreDaysOfReads stores a batch of DayOfReads elements. It is a optimized operation in that:
//    1. One element represents a relatively short-and-wide entry of all reads for a single day.
//    2. We have multiple DayOfReads elements and we use a PutMulti to make this faster.
// For details of how a single element of DayOfReads is physically stored, see the implementation of apimodel.Store and apimodel.Load.
// Also important to note, this store operation also handles updating the GlukitUser entry with the most recent read, if applicable.
func StoreDaysOfReads(context appengine.Context, userProfileKey *datastore.Key, daysOfReads []apimodel.DayOfGlucoseReads) (keys []*datastore.Key, err error) {
	elementKeys := make([]*datastore.Key, len(daysOfReads))
	for i := range daysOfReads {
		elementKeys[i] = datastore.NewKey(context, "DayOfReads", "", daysOfReads[i].Reads[0].Time.Timestamp, userProfileKey)
	}

	context.Infof("Emitting a PutMulti with %d keys for all %d days of reads", len(elementKeys), len(daysOfReads))
	keys, error := datastore.PutMulti(context, elementKeys, daysOfReads)
	if error != nil {
		context.Warningf("Error writing %d days of reads with keys [%s]: %v", len(elementKeys), elementKeys, error)
		return nil, error
	}

	// Get the time of the batch's last read and update the most recent read timestamp if necessary
	userProfile, err := GetGlukitUserWithKey(context, userProfileKey)
	if err != nil {
		context.Criticalf("Error reading user profile [%s] for its most recent read value: %v", userProfileKey, err)
		return nil, err
	}

	lastDayOfRead := daysOfReads[len(daysOfReads)-1]
	lastRead := lastDayOfRead.Reads[len(lastDayOfRead.Reads)-1]
	if userProfile.MostRecentRead.GetTime().Before(lastRead.GetTime()) {
		context.Infof("Updating most recent read date to %s", lastRead.GetTime())
		userProfile.MostRecentRead = lastRead
		_, err := StoreUserProfile(context, time.Now(), *userProfile)
		if err != nil {
			context.Criticalf("Error storing updated user profile [%s] with most recent read value of %s: %v", userProfileKey, userProfile.MostRecentRead, err)
			return nil, err
		}
	}

	return elementKeys, nil
}

// StoreCalibrationReads stores a batch of DayOfCalibrations elements. It is a optimized operation in that:
//    1. One element represents a relatively short-and-wide entry of all calibration reads for a single day.
//    2. We have multiple DayOfReads elements and we use a PutMulti to make this faster.
// For details of how a single element of DayOfReads is physically stored, see the implementation of apimodel.Store and apimodel.Load.
func StoreCalibrationReads(context appengine.Context, userProfileKey *datastore.Key, daysOfCalibrationReads []apimodel.DayOfCalibrationReads) (keys []*datastore.Key, err error) {
	elementKeys := make([]*datastore.Key, len(daysOfCalibrationReads))
	for i := range daysOfCalibrationReads {
		elementKeys[i] = datastore.NewKey(context, "DayOfCalibrationReads", "", daysOfCalibrationReads[i].Reads[0].Time.Timestamp, userProfileKey)
	}

	context.Infof("Emitting a PutMulti with %d keys for all %d days of calibration reads", len(elementKeys), len(daysOfCalibrationReads))
	keys, error := datastore.PutMulti(context, elementKeys, daysOfCalibrationReads)
	if error != nil {
		context.Criticalf("Error writing %d days of calibration reads with keys [%s]: %v", len(elementKeys), elementKeys, error)
		return nil, error
	}

	return elementKeys, nil
}

// GetGlucoseReads returns all GlucoseReads given a user's email address and the time boundaries. Not that the boundaries are both inclusive.
func GetGlucoseReads(context appengine.Context, email string, lowerBound time.Time, upperBound time.Time) (reads []apimodel.GlucoseRead, err error) {
	key := GetUserKey(context, email)

	// Scan start should be one day prior and scan end should be one day later so that we can capture the day using
	// a single column inequality filter. The scan should actually capture at least one day and a maximum of 3
	scanStart := lowerBound.Add(time.Duration(-24 * time.Hour))
	scanEnd := upperBound.Add(time.Duration(24 * time.Hour))

	context.Infof("Scanning for reads between %s and %s to get reads between %s and %s", scanStart, scanEnd, lowerBound, upperBound)

	query := datastore.NewQuery("DayOfReads").Ancestor(key).Filter("startTime >=", scanStart).Filter("startTime <=", scanEnd).Order("startTime")
	daysOfReads := new(apimodel.DayOfGlucoseReads)
	readsForPeriod := make([]apimodel.GlucoseRead, 0)

	iterator := query.Run(context)
	for _, err := iterator.Next(daysOfReads); err == nil; _, err = iterator.Next(daysOfReads) {
		context.Debugf("Loaded batch of %d reads...", len(daysOfReads.Reads))
		readsForPeriod = mergeGlucoseReadArrays(readsForPeriod, daysOfReads.Reads)
		daysOfReads = new(apimodel.DayOfGlucoseReads)
	}
{'fraction_non_alphanumeric': 0.09745552484229925, 'fraction_numerical': 0.0024098093415550357, 'mean_word_length': 4.986423419601188, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4508631', 'n_tokens_mistral': 4498, 'n_tokens_neox': 4316, 'n_words': 1458}