service.h

// Copyright 2025 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
 
#ifndef ACTIONENGINE_SERVICE_SERVICE_H_
#define ACTIONENGINE_SERVICE_SERVICE_H_
 
#include <functional>
#include <memory>
#include <string>
#include <string_view>
#include <vector>
 
#include <absl/base/nullability.h>
#include <absl/base/thread_annotations.h>
#include <absl/container/flat_hash_map.h>
#include <absl/container/flat_hash_set.h>
#include <absl/status/status.h>
#include <absl/status/statusor.h>
 
#include "actionengine/actions/action.h"
#include "actionengine/actions/registry.h"
#include "actionengine/concurrency/concurrency.h"
#include "actionengine/net/stream.h"
#include "actionengine/nodes/node_map.h"
#include "actionengine/service/session.h"
#include "actionengine/stores/chunk_store.h"
 
namespace act {

struct StreamToSessionConnection {
  std::shared_ptr<WireStream> stream = nullptr;
  Session* absl_nullable session = nullptr;
 
  std::string session_id;  // dead sessions may lose their id.
  std::string stream_id;   // dead streams may lose their id.
 
  absl::Status status;
};
 
std::unique_ptr<Action> MakeActionInConnection(
    const StreamToSessionConnection& connection, std::string_view action_name,
    std::string_view action_id = "");

using ConnectionHandler = std::function<absl::Status(
    const std::shared_ptr<WireStream>&, Session* absl_nonnull)>;

absl::Status RunSimpleSession(std::shared_ptr<WireStream> stream,
                              Session* absl_nonnull session);

class Service : public std::enable_shared_from_this<Service> {
 public:
  explicit Service(ActionRegistry* absl_nullable action_registry = nullptr,
                   ConnectionHandler connection_handler = RunSimpleSession,
                   ChunkStoreFactory chunk_store_factory = {});
 
  ~Service();
 
  // This class is not copyable or movable.
  Service(const Service& other) = delete;
  Service& operator=(const Service& other) = delete;

  auto GetStream(std::string_view stream_id) const -> WireStream* absl_nullable;
  auto GetSession(std::string_view session_id) const -> Session* absl_nullable;
  auto GetSessionKeys() const -> std::vector<std::string>;

  auto EstablishConnection(std::shared_ptr<WireStream>&& stream,
                           ConnectionHandler connection_handler = nullptr)
      -> absl::StatusOr<std::shared_ptr<StreamToSessionConnection>>;
  auto JoinConnection(StreamToSessionConnection* absl_nonnull connection)
      -> absl::Status;

  auto SetActionRegistry(const ActionRegistry& action_registry) const -> void;
 
  void JoinConnectionsAndCleanUp(bool cancel = false) ABSL_LOCKS_EXCLUDED(mu_);
 
 private:
  void CleanupConnection(const StreamToSessionConnection& connection)
      ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_);
 
  std::unique_ptr<ActionRegistry> action_registry_;
  ConnectionHandler connection_handler_;
  ChunkStoreFactory chunk_store_factory_;
 
  mutable act::Mutex mu_;
  absl::flat_hash_map<std::string, std::shared_ptr<WireStream>> streams_
      ABSL_GUARDED_BY(mu_);
  // for now, we only support one-to-one session-stream mapping, therefore we
  // use the stream id as the session id.
  absl::flat_hash_map<std::string, std::unique_ptr<NodeMap>> node_maps_
      ABSL_GUARDED_BY(mu_);
  absl::flat_hash_map<std::string, std::unique_ptr<Session>> sessions_
      ABSL_GUARDED_BY(mu_);
  absl::flat_hash_map<std::string, std::shared_ptr<StreamToSessionConnection>>
      connections_ ABSL_GUARDED_BY(mu_);
  absl::flat_hash_map<std::string, absl::flat_hash_set<std::string>>
      streams_per_session_ ABSL_GUARDED_BY(mu_);
  absl::flat_hash_map<std::string, std::unique_ptr<thread::Fiber>>
      connection_fibers_ ABSL_GUARDED_BY(mu_);
 
  bool cleanup_started_ ABSL_GUARDED_BY(mu_) = false;
  thread::PermanentEvent cleanup_done_;
};
 
}  // namespace act
 
#endif  // ACTIONENGINE_SERVICE_SERVICE_H_