From c856da153f8c5b165cdc46a22d06cb6f2ca7e4b5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Michael=20He=C3=9F?= Date: Thu, 26 Oct 2023 20:17:21 +0200 Subject: [PATCH] Session log: handle unfinished sessions (#10328) --- core/loadpoint.go | 2 +- core/loadpoint_session_test.go | 96 ++++++++++++++++++++++++++++++++++ core/session/db.go | 35 +++++++++++-- core/site.go | 4 ++ 4 files changed, 131 insertions(+), 6 deletions(-) diff --git a/core/loadpoint.go b/core/loadpoint.go index f88e9d096..9e9f1844d 100644 --- a/core/loadpoint.go +++ b/core/loadpoint.go @@ -175,7 +175,7 @@ type Loadpoint struct { progress *Progress // Step-wise progress indicator // session log - db session.Database + db *session.DB session *session.Session tasks *util.Queue[Task] // tasks to be executed diff --git a/core/loadpoint_session_test.go b/core/loadpoint_session_test.go index aeb637497..fdc9a1e7d 100644 --- a/core/loadpoint_session_test.go +++ b/core/loadpoint_session_test.go @@ -87,3 +87,99 @@ func TestSession(t *testing.T) { assert.Len(t, s, 1) t.Logf("session: %+v", s) } + +func TestCloseSessionsOnStartup_emptyDb(t *testing.T) { + var err error + serverdb.Instance, err = serverdb.New("sqlite", ":memory:") + assert.NoError(t, err) + + db, err := session.NewStore("foo", serverdb.Instance) + assert.NoError(t, err) + + // assert empty DB is no problem + err = db.ClosePendingSessionsInHistory(1000) + assert.NoError(t, err) +} + +func TestCloseSessionsOnStartup(t *testing.T) { + var err error + serverdb.Instance, err = serverdb.New("sqlite", ":memory:") + assert.NoError(t, err) + + db1, err := session.NewStore("foo", serverdb.Instance) + assert.NoError(t, err) + + db2, err := session.NewStore("bar", serverdb.Instance) + assert.NoError(t, err) + + clock := clock.NewMock() + + //test data, creates 6 sessions for each loadpoint, 3rd and 6th are "unfinished" + var sessions1 []*session.Session = createMockSessions(db1, clock) + var sessions2 []*session.Session = createMockSessions(db2, clock) + + //write interleaved for two loadpoints + for index, session := range sessions1 { + db1.Persist(session) + db2.Persist(sessions2[index]) + } + + err = db1.ClosePendingSessionsInHistory(1000) + assert.NoError(t, err) + + //check fixed sessions for db1 + var db1Sessions session.Sessions + err = serverdb.Instance.Where("Loadpoint = ?", "foo").Order("ID").Find(&db1Sessions).Error + assert.NoError(t, err) + assert.Len(t, db1Sessions, 6) + + //check fixed history + for _, s := range db1Sessions[:5] { + assert.NotEmpty(t, s.MeterStop) + assert.Equal(t, float64(10), s.ChargedEnergy) + t.Logf("session: %+v", s) + } + + //check fixed most recent record + assert.NotEmpty(t, db1Sessions[5].MeterStop) + assert.Equal(t, float64(940), db1Sessions[5].ChargedEnergy) + + //ensure no side effects on loadpoint 2 data, i.e. data left unfixed + var db2Sessions session.Sessions + err = serverdb.Instance.Where("Loadpoint = ?", "bar").Order("ID").Find(&db2Sessions).Error + assert.NoError(t, err) + assert.Len(t, db2Sessions, 6) + + for i, s := range db2Sessions { + if (i+1)%3 == 0 { + assert.Empty(t, s.MeterStop) + assert.Empty(t, s.ChargedEnergy) + continue + } + assert.NotEmpty(t, s.MeterStop) + assert.Equal(t, float64(10), s.ChargedEnergy) + } +} + +func createMockSessions(db *session.DB, clock *clock.Mock) []*session.Session { + var sessions []*session.Session + for i := 1; i <= 6; i++ { + + var meter1Start float64 = float64(i * 10) + session := db.New(meter1Start) + session.Created = clock.Now().Add(1 * time.Minute) + + //create every third session as incomplete + if i%3 == 0 { + sessions = append(sessions, session) + continue + } + + session.Finished = clock.Now().Add(2 * time.Minute) + meterStop := float64(meter1Start + 10) + session.MeterStop = &meterStop + session.ChargedEnergy = 10 + sessions = append(sessions, session) + } + return sessions +} diff --git a/core/session/db.go b/core/session/db.go index 9c3e70a8b..57ad1f308 100644 --- a/core/session/db.go +++ b/core/session/db.go @@ -12,11 +12,6 @@ type DB struct { name string } -type Database interface { - New(startEnergy float64) *Session - Persist(session interface{}) -} - // NewStore creates a session store func NewStore(name string, db *gorm.DB) (*DB, error) { err := db.AutoMigrate(new(Session)) @@ -57,3 +52,33 @@ func (s *DB) Sessions() (Sessions, error) { tx := s.db.Find(&res) return res, tx.Error } + +func (s *DB) ClosePendingSessionsInHistory(chargeMeterTotal float64) error { + var res Sessions + if tx := s.db.Find(&res, map[string]interface{}{"finished": "0001-01-01 00:00:00+00:00", "Loadpoint": s.name}); tx.Error != nil { + return tx.Error + } + + for _, session := range res { + var nextSession Session + + var tx *gorm.DB + if tx = s.db.Limit(1).Order("ID").Find(&nextSession, "ID > ? AND Loadpoint = ?", session.ID, s.name); tx.Error != nil { + return tx.Error + } + + if tx.RowsAffected == 0 { + // no successor, this is the most recent session and it is open + session.MeterStop = &chargeMeterTotal + } else { + session.MeterStop = nextSession.MeterStart + } + + if session.MeterStart != nil && session.MeterStop != nil { + session.ChargedEnergy = *session.MeterStop - *session.MeterStart + s.Persist(session) + } + } + + return nil +} diff --git a/core/site.go b/core/site.go index 8cf3cd361..786075745 100644 --- a/core/site.go +++ b/core/site.go @@ -142,6 +142,10 @@ func NewSiteFromConfig( if lp.db, err = session.NewStore(lp.Title(), db.Instance); err != nil { return nil, err } + // Fix any dangling history + if err := lp.db.ClosePendingSessionsInHistory(lp.chargeMeterTotal()); err != nil { + return nil, err + } // NOTE: this requires stopSession to respect async access shutdown.Register(lp.stopSession)