diff --git a/app/sephirah/internal/biz/biznetzach/netzach.go b/app/sephirah/internal/biz/biznetzach/netzach.go index 4bf7d55..159717c 100644 --- a/app/sephirah/internal/biz/biznetzach/netzach.go +++ b/app/sephirah/internal/biz/biznetzach/netzach.go @@ -165,6 +165,18 @@ func (n *Netzach) CreateNotifyFlow(ctx context.Context, flow *modelnetzach.Notif if err != nil { return 0, pb.ErrorErrorReasonUnspecified("%s", err.Error()) } + if len(flow.Sources) > 0 { + for _, source := range flow.Sources { + err = n.notifySourceCache.Delete(ctx, source.SourceID) + if err != nil { + logger.Errorf("failed to delete cache %s", err.Error()) + } + } + } + err = n.notifyFlowCache.Delete(ctx, flow.ID) + if err != nil { + logger.Errorf("failed to delete cache %s", err.Error()) + } return flow.ID, nil } diff --git a/app/sephirah/internal/data/netzach.go b/app/sephirah/internal/data/netzach.go index f05e278..9266150 100644 --- a/app/sephirah/internal/data/netzach.go +++ b/app/sephirah/internal/data/netzach.go @@ -10,7 +10,6 @@ import ( "github.com/tuihub/librarian/app/sephirah/internal/data/internal/ent/notifyflow" "github.com/tuihub/librarian/app/sephirah/internal/data/internal/ent/notifyflowsource" "github.com/tuihub/librarian/app/sephirah/internal/data/internal/ent/notifyflowtarget" - "github.com/tuihub/librarian/app/sephirah/internal/data/internal/ent/notifysource" "github.com/tuihub/librarian/app/sephirah/internal/data/internal/ent/notifytarget" "github.com/tuihub/librarian/app/sephirah/internal/data/internal/ent/systemnotification" "github.com/tuihub/librarian/app/sephirah/internal/data/internal/ent/user" @@ -264,7 +263,7 @@ func (n *netzachRepo) GetNotifyFlow(ctx context.Context, id model.InternalID) (* func (n *netzachRepo) GetNotifyFlowIDsWithFeed(ctx context.Context, id model.InternalID) ([]model.InternalID, error) { ids, err := n.data.db.NotifyFlow.Query().Where( - notifyflow.HasNotifySourceWith(notifysource.FeedConfigIDEQ(id)), + notifyflow.HasNotifyFlowSourceWith(notifyflowsource.NotifySourceIDEQ(id)), ).IDs(ctx) if err != nil { return nil, err diff --git a/app/sephirah/internal/supervisor/supervisor.go b/app/sephirah/internal/supervisor/supervisor.go index 4aa71af..ac19399 100644 --- a/app/sephirah/internal/supervisor/supervisor.go +++ b/app/sephirah/internal/supervisor/supervisor.go @@ -348,28 +348,35 @@ func (s *Supervisor) updateControllers( ctl.ConnectionStatus = modelsupervisor.PorterConnectionStatusActive ctl.LastHeartbeat = now } - if resp == nil || resp.GetEnablesSummary() == nil { + if ctl.ConnectionStatus == modelsupervisor.PorterConnectionStatusActive && resp == nil { return nil } + var reQueueContext []model.InternalID if ctl.ConnectionStatus == modelsupervisor.PorterConnectionStatusActive { onlyLast, _ := libtype.DiffSlices( ctl.LastEnabledContext, converter.ToBizInternalIDList(resp.GetEnablesSummary().GetContextIds()), ) + reQueueContext = onlyLast ctl.LastEnabledContext = converter.ToBizInternalIDList(resp.GetEnablesSummary().GetContextIds()) - for _, id := range onlyLast { - ic, ok := s.instanceContextController.Load(id) - if !ok { - continue - } - ic.HandleStatus = modelsupervisor.PorterContextHandleStatusQueueing - ic.HandleStatusMessage = "" - ic.HandlerAddress = "" - s.instanceContextController.Store(id, ic) - err := s.QueuePorterContext(ctx, ic.PorterContext) - if err != nil { - return err - } + } else { + reQueueContext = ctl.LastEnabledContext + if ctl.ConnectionStatus == modelsupervisor.PorterConnectionStatusDowngraded { + ctl.LastEnabledContext = nil + } + } + for _, id := range reQueueContext { + ic, ok := s.instanceContextController.Load(id) + if !ok { + continue + } + ic.HandleStatus = modelsupervisor.PorterContextHandleStatusQueueing + ic.HandleStatusMessage = "" + ic.HandlerAddress = "" + s.instanceContextController.Store(id, ic) + err := s.QueuePorterContext(ctx, ic.PorterContext) + if err != nil { + return err } } return nil