曾志威
9da95beb42
continuous-integration/drone Build is failing
Migrate the workspace to React and Ant Design, improve account forms and responsive layouts, support outbound diagnostics and SSH fingerprint confirmation, and persist the management startup port. Preserve database migrations, remove S3 settings storage, and improve gateway audit handling.
123 lines
3.1 KiB
Go
123 lines
3.1 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"linkport/internal/model"
|
|
"log"
|
|
"time"
|
|
)
|
|
|
|
type EventFilter struct {
|
|
Gateway, Source, Category, Keyword, From, To string
|
|
Scope model.ResourceScope
|
|
}
|
|
|
|
func (s *Store) Event(id, source, category, detail string) {
|
|
s.auditOps.Lock()
|
|
defer s.auditOps.Unlock()
|
|
at := time.Now().UTC()
|
|
db, err := s.auditDBForWrite(at.Format("2006-01-02"), int64(len(id)+len(source)+len(category)+len(detail)+512))
|
|
if err == nil {
|
|
_, err = db.Exec("INSERT INTO events(proxy_id,source,category,detail,at) VALUES(?,?,?,?,?)", id, source, category, detail, at.Format(time.RFC3339Nano))
|
|
}
|
|
if err != nil {
|
|
log.Print("cannot persist connection event")
|
|
}
|
|
}
|
|
func (s *Store) Events() ([]Event, error) {
|
|
return s.FilterEvents(context.Background(), EventFilter{}, 0)
|
|
}
|
|
func (s *Store) FilterEvents(ctx context.Context, f EventFilter, before int64) ([]Event, error) {
|
|
out := []Event{}
|
|
err := s.WalkEvents(ctx, f, before, 300, func(e Event) error { out = append(out, e); return nil })
|
|
return out, err
|
|
}
|
|
func (s *Store) WalkEvents(ctx context.Context, f EventFilter, before int64, limit int, visit func(Event) error) error {
|
|
var scopes ScopeIndex
|
|
if f.Scope != (model.ResourceScope{}) {
|
|
var err error
|
|
scopes, err = s.ResourceScopes(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
s.auditOps.RLock()
|
|
defer s.auditOps.RUnlock()
|
|
days, err := s.auditDays()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
bases := auditShardBases(days)
|
|
count := 0
|
|
for _, day := range days {
|
|
base := bases[day]
|
|
if before > 0 && base >= before {
|
|
continue
|
|
}
|
|
cursor := int64(0)
|
|
if before > base && before < base+auditShardStride {
|
|
cursor = before - base
|
|
}
|
|
query := "SELECT id,proxy_id,source,category,detail,at FROM events WHERE (?=0 OR id<?)"
|
|
args := []any{cursor, cursor}
|
|
for _, item := range [][2]string{{"proxy_id", f.Gateway}, {"source", f.Source}, {"category", f.Category}} {
|
|
if item[1] != "" {
|
|
query += " AND " + item[0] + "=?"
|
|
args = append(args, item[1])
|
|
}
|
|
}
|
|
if f.Keyword != "" {
|
|
query += " AND instr(lower(detail),lower(?))>0"
|
|
args = append(args, f.Keyword)
|
|
}
|
|
if f.From != "" {
|
|
query += " AND julianday(at)>=julianday(?)"
|
|
args = append(args, f.From)
|
|
}
|
|
if f.To != "" {
|
|
query += " AND julianday(at)<=julianday(?)"
|
|
args = append(args, f.To)
|
|
}
|
|
query += " ORDER BY id DESC"
|
|
db, e := s.auditDB(day)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
rows, e := db.QueryContext(ctx, query, args...)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
for rows.Next() {
|
|
var event Event
|
|
if e = rows.Scan(&event.ID, &event.ProxyID, &event.Source, &event.Category, &event.Detail, &event.At); e != nil {
|
|
rows.Close()
|
|
return e
|
|
}
|
|
if event.ID >= auditShardStride {
|
|
rows.Close()
|
|
return errors.New("event shard row limit exceeded")
|
|
}
|
|
if scopes != nil && !scopes.MatchesResource(event.ProxyID, f.Scope) {
|
|
continue
|
|
}
|
|
event.ID += base
|
|
if e = visit(event); e != nil {
|
|
rows.Close()
|
|
return e
|
|
}
|
|
count++
|
|
if limit > 0 && count >= limit {
|
|
rows.Close()
|
|
return nil
|
|
}
|
|
}
|
|
e = rows.Err()
|
|
rows.Close()
|
|
if e != nil {
|
|
return e
|
|
}
|
|
}
|
|
return nil
|
|
}
|