124 lines
3.3 KiB
Go
124 lines
3.3 KiB
Go
package store
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"os"
|
|
"path/filepath"
|
|
|
|
"git.ipao.vip/rogee/go-sip/internal/contract"
|
|
)
|
|
|
|
type SIPSource struct {
|
|
Original []byte
|
|
Generation int64
|
|
name string
|
|
}
|
|
|
|
func (s *Store) sipSourceDir() string {
|
|
return filepath.Join(filepath.Dir(s.path), "sip-notification-sources")
|
|
}
|
|
|
|
// A valid notification is acknowledged before SIP can be loaded when old calls
|
|
// must drain. Keep its exact source until the correlated read/apply completes.
|
|
func (s *Store) RememberSIPSource(dispatcherID string, original []byte) error {
|
|
if err := contract.ValidateCurrent("mq", original); err != nil {
|
|
return err
|
|
}
|
|
var id struct {
|
|
EventType string `json:"event_type"`
|
|
DispatcherID string `json:"dispatcher_id"`
|
|
Payload struct {
|
|
Status string `json:"status"`
|
|
} `json:"payload"`
|
|
}
|
|
if json.Unmarshal(original, &id) != nil || id.EventType != "sip.config" || id.DispatcherID != dispatcherID || id.Payload.Status != "" {
|
|
return ErrUnidentifiableData
|
|
}
|
|
_, generation, err := s.SIPState(dispatcherID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if generation <= 0 {
|
|
return errors.New("SIP source requires a pending internal generation")
|
|
}
|
|
data, err := json.Marshal(SIPSource{Original: original, Generation: generation})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
sum := sha256.Sum256(data)
|
|
return persistPrivateData(s.sipSourceDir(), hex.EncodeToString(sum[:])+".json", data)
|
|
}
|
|
func (s *Store) PendingSIPSources() ([]SIPSource, error) {
|
|
entries, err := os.ReadDir(s.sipSourceDir())
|
|
if errors.Is(err, os.ErrNotExist) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var sources []SIPSource
|
|
for _, entry := range entries {
|
|
name := filepath.Join(s.sipSourceDir(), entry.Name())
|
|
info, err := os.Lstat(name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !info.Mode().IsRegular() || info.Mode().Perm() != 0600 {
|
|
return nil, errors.New("SIP notification source is not private")
|
|
}
|
|
raw, err := os.ReadFile(name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
sum := sha256.Sum256(raw)
|
|
if filepath.Base(name) != hex.EncodeToString(sum[:])+".json" {
|
|
return nil, errors.New("SIP notification original bytes changed")
|
|
}
|
|
var source SIPSource
|
|
if err := json.Unmarshal(raw, &source); err != nil {
|
|
return nil, err
|
|
}
|
|
if source.Generation <= 0 {
|
|
return nil, errors.New("SIP source generation missing")
|
|
}
|
|
source.name = filepath.Base(name)
|
|
sources = append(sources, source)
|
|
}
|
|
return sources, nil
|
|
}
|
|
func (s *Store) CompleteAppliedSIPSources(dispatcherID string) error {
|
|
var applied int64
|
|
if err := s.db.QueryRow(`SELECT applied_sip_revision FROM dispatcher_state WHERE dispatcher_id=?`, dispatcherID).Scan(&applied); err != nil {
|
|
return err
|
|
}
|
|
sources, err := s.PendingSIPSources()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, source := range sources {
|
|
if source.Generation <= applied {
|
|
if err := s.CompleteSIPSource(source); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
func (s *Store) CompleteSIPSource(source SIPSource) error {
|
|
data, err := json.Marshal(source)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
sum := sha256.Sum256(data)
|
|
if source.name != hex.EncodeToString(sum[:])+".json" {
|
|
return errors.New("invalid SIP notification source identity")
|
|
}
|
|
if err := os.Remove(filepath.Join(s.sipSourceDir(), source.name)); err != nil {
|
|
return err
|
|
}
|
|
return syncPrivateDirectory(s.sipSourceDir())
|
|
}
|