diff --git a/Cargo.lock b/Cargo.lock
index 0c2142d132..fb6be2b383 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7187,6 +7187,7 @@ dependencies = [
  "async-recursion",
  "async-stream",
  "async-trait",
+ "audit",
  "aws-sdk-sns",
  "axum",
  "axum-client-ip",
@@ -7245,6 +7246,7 @@ dependencies = [
  "openobserve-core",
  "openobserve-jobs",
  "openobserve-mcp",
+ "openobserve-node",
  "opentelemetry",
  "opentelemetry-otlp",
  "opentelemetry-proto",
@@ -7307,6 +7309,7 @@ dependencies = [
  "transform",
  "uaparser",
  "url",
+ "usage_reporting",
  "utoipa",
  "utoipa-swagger-ui",
  "version-compare",
@@ -7365,7 +7368,9 @@ dependencies = [
  "openobserve-cipher",
  "openobserve-core",
  "openobserve-mcp",
+ "openobserve-node",
  "openobserve-sourcemaps",
+ "openobserve-tls",
  "opentelemetry",
  "opentelemetry-proto",
  "parking_lot 0.12.5",
@@ -7395,6 +7400,7 @@ dependencies = [
  "tracing-opentelemetry",
  "transform",
  "url",
+ "usage_reporting",
  "utoipa",
  "utoipa-swagger-ui",
  "zip 8.2.0",
@@ -7436,6 +7442,7 @@ dependencies = [
  "o2_openfga",
  "openobserve-api-common",
  "openobserve-core",
+ "openobserve-node",
  "openobserve-org-storage",
  "regex",
  "reqwest 0.13.2",
@@ -7450,6 +7457,7 @@ dependencies = [
  "tracing",
  "transform",
  "url",
+ "usage_reporting",
  "utoipa",
  "vrl",
 ]
@@ -7565,6 +7573,7 @@ dependencies = [
  "o2_dex",
  "o2_enterprise",
  "o2_openfga",
+ "openobserve-node",
  "opentelemetry",
  "opentelemetry-proto",
  "parking_lot 0.12.5",
@@ -7576,11 +7585,8 @@ dependencies = [
  "prost-wkt-types",
  "proto",
  "rand 0.10.1",
- "rcgen",
  "regex",
  "reqwest 0.13.2",
- "rustls",
- "rustls-native-certs",
  "schema",
  "sea-orm",
  "search",
@@ -7603,8 +7609,6 @@ dependencies = [
  "utoipa",
  "version-compare",
  "vrl",
- "webpki-roots 1.0.6",
- "x509-parser",
 ]
 
 [[package]]
@@ -7634,6 +7638,7 @@ dependencies = [
  "o2_openfga",
  "openobserve-builtins",
  "openobserve-core",
+ "openobserve-node",
  "openobserve-org-storage",
  "prometheus",
  "promql-service",
@@ -7650,6 +7655,7 @@ dependencies = [
  "tokio",
  "tonic",
  "transform",
+ "usage_reporting",
  "wal",
 ]
 
@@ -7672,6 +7678,22 @@ dependencies = [
  "utoipa",
 ]
 
+[[package]]
+name = "openobserve-node"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "common",
+ "config",
+ "db",
+ "infra",
+ "log",
+ "prometheus",
+ "proto",
+ "tokio",
+ "tonic",
+]
+
 [[package]]
 name = "openobserve-org-storage"
 version = "0.1.0"
@@ -7713,6 +7735,22 @@ dependencies = [
  "zip 8.2.0",
 ]
 
+[[package]]
+name = "openobserve-tls"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "config",
+ "itertools 0.14.0",
+ "log",
+ "rcgen",
+ "reqwest 0.13.2",
+ "rustls",
+ "rustls-native-certs",
+ "webpki-roots 1.0.6",
+ "x509-parser",
+]
+
 [[package]]
 name = "openssl-probe"
 version = "0.2.1"
@@ -10138,13 +10176,16 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "arrow-schema",
+ "async-trait",
  "common",
  "config",
  "dashmap",
  "db",
  "hashbrown 0.16.1",
  "infra",
+ "ingester",
  "ingestion-common",
+ "itertools 0.14.0",
  "log",
  "serde_json",
  "tokio",
diff --git a/Cargo.toml b/Cargo.toml
index 8221aad302..58c0d9d4d7 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -24,12 +24,14 @@ publish = false
 [features]
 default = []
 enterprise = [
+    "audit/enterprise",
     "openobserve-api/enterprise",
     "openobserve-core/enterprise",
     "openobserve-jobs/enterprise",
     "super_cluster_queue/enterprise",
     "search_service/enterprise",
     "enrichment-data/enterprise",
+    "usage_reporting/enterprise",
 ]
 vectorscan = [
     "openobserve-api/vectorscan",
@@ -215,10 +217,12 @@ pin-project-lite.workspace = true
 flight.workspace = true
 async-stream.workspace = true
 sourcemap.workspace = true
+audit.workspace = true
 openobserve-api.workspace = true
 web.workspace = true
 openobserve-api-management.workspace = true
 openobserve-core.workspace = true
+openobserve-node.workspace = true
 ingestion-common.workspace = true
 compaction.workspace = true
 enrichment-data.workspace = true
@@ -227,6 +231,7 @@ db.workspace = true
 search.workspace = true
 search_service.workspace = true
 super_cluster_queue.workspace = true
+usage_reporting.workspace = true
 openobserve-jobs.workspace = true
 openobserve-mcp.workspace = true
 
@@ -259,8 +264,10 @@ members = [
     "src/builtins",
     "src/org_storage",
     "src/sourcemaps",
+    "src/tls",
     "src/common",
     "src/core",
+    "src/node",
     "src/transform",
     "src/super_cluster_queue",
     "src/search",
@@ -307,8 +314,10 @@ enrichment-data = { path = "src/enrichment_data" }
 openobserve-builtins = { path = "src/builtins" }
 openobserve-org-storage = { path = "src/org_storage" }
 openobserve-sourcemaps = { path = "src/sourcemaps" }
+openobserve-tls = { path = "src/tls" }
 web = { path = "src/web" }
 openobserve-core = { path = "src/core" }
+openobserve-node = { path = "src/node" }
 common = { path = "src/common" }
 transform = { path = "src/transform" }
 super_cluster_queue = { path = "src/super_cluster_queue" }
diff --git a/deny.toml b/deny.toml
index c1ab32f29c..59bc1bf7ea 100644
--- a/deny.toml
+++ b/deny.toml
@@ -30,10 +30,12 @@ exceptions = [
     { name = "openobserve-builtins", allow = ["AGPL-3.0"] },
     { name = "common", allow = ["AGPL-3.0"] },
     { name = "openobserve-core", allow = ["AGPL-3.0"] },
+    { name = "openobserve-node", allow = ["AGPL-3.0"] },
     { name = "openobserve-jobs", allow = ["AGPL-3.0"] },
     { name = "openobserve-mcp", allow = ["AGPL-3.0"] },
     { name = "openobserve-org-storage", allow = ["AGPL-3.0"] },
     { name = "openobserve-sourcemaps", allow = ["AGPL-3.0"] },
+    { name = "openobserve-tls", allow = ["AGPL-3.0"] },
     { name = "transform", allow = ["AGPL-3.0"] },
     { name = "super_cluster_queue", allow = ["AGPL-3.0"] },
     { name = "search", allow = ["AGPL-3.0"] },
diff --git a/src/api/Cargo.toml b/src/api/Cargo.toml
index 62f08a4764..2df98275ab 100644
--- a/src/api/Cargo.toml
+++ b/src/api/Cargo.toml
@@ -25,6 +25,7 @@ enterprise = [
     "search_service/enterprise",
     "enrichment-data/enterprise",
     "stream/enterprise",
+    "usage_reporting/enterprise",
 ]
 cloud = [
     "enterprise",
@@ -58,8 +59,10 @@ common.workspace = true
 audit.workspace = true
 db.workspace = true
 openobserve-core.workspace = true
+openobserve-node.workspace = true
 openobserve-builtins.workspace = true
 openobserve-sourcemaps.workspace = true
+openobserve-tls.workspace = true
 ingestion-common.workspace = true
 stream.workspace = true
 enrichment-data.workspace = true
@@ -71,6 +74,7 @@ openobserve-api-management.workspace = true
 openobserve-api-query.workspace = true
 openobserve-api-common.workspace = true
 openobserve-mcp.workspace = true
+usage_reporting.workspace = true
 axum.workspace = true
 axum-extra.workspace = true
 tower.workspace = true
diff --git a/src/api/src/handler/http/request/stream/mod.rs b/src/api/src/handler/http/request/stream/mod.rs
index 3b9c4aac4c..74e939176b 100644
--- a/src/api/src/handler/http/request/stream/mod.rs
+++ b/src/api/src/handler/http/request/stream/mod.rs
@@ -949,7 +949,7 @@ async fn get_super_cluster_delete_status(
     let mut errors = Vec::new();
 
     for cluster in clusters {
-        match openobserve_core::cluster_info::get_super_cluster_delete_job_status(
+        match openobserve_node::cluster_info::get_super_cluster_delete_job_status(
             &trace_id,
             cluster.clone(),
             id,
diff --git a/src/api/src/handler/http/request/synthetics/mod.rs b/src/api/src/handler/http/request/synthetics/mod.rs
index 739925451b..491583e661 100644
--- a/src/api/src/handler/http/request/synthetics/mod.rs
+++ b/src/api/src/handler/http/request/synthetics/mod.rs
@@ -1091,21 +1091,19 @@ async fn process_ack(
     let resp = o2_enterprise::enterprise::synthetics::job_api::ack(req, token_org).await?;
 
     // Emit trigger usage record for synthetics telemetry.
-    openobserve_core::self_reporting::publish_triggers_usage(
-        config::meta::self_reporting::usage::TriggerData {
-            _timestamp: checked_at,
-            org: resp.org_id.clone(),
-            module: config::meta::self_reporting::usage::TriggerDataType::Synthetics,
-            key: format!("{}/{}", resp.synthetics_name, resp.synthetics_id),
-            start_time: checked_at,
-            end_time: checked_at,
-            status: config::meta::self_reporting::usage::TriggerDataStatus::Completed,
-            success_response: Some(status.clone()),
-            error: error.clone(),
-            evaluation_took_in_secs: Some(response_time_ms / 1000.0),
-            ..Default::default()
-        },
-    );
+    usage_reporting::publish_triggers_usage(config::meta::self_reporting::usage::TriggerData {
+        _timestamp: checked_at,
+        org: resp.org_id.clone(),
+        module: config::meta::self_reporting::usage::TriggerDataType::Synthetics,
+        key: format!("{}/{}", resp.synthetics_name, resp.synthetics_id),
+        start_time: checked_at,
+        end_time: checked_at,
+        status: config::meta::self_reporting::usage::TriggerDataStatus::Completed,
+        success_response: Some(status.clone()),
+        error: error.clone(),
+        evaluation_took_in_secs: Some(response_time_ms / 1000.0),
+        ..Default::default()
+    });
 
     // Notify once per run, not once per job ack.
     if resp.run_complete && !resp.destinations.is_empty() {
diff --git a/src/api/src/router/grpc/ingest/logs.rs b/src/api/src/router/grpc/ingest/logs.rs
index 84dd4cc20e..e839f3fc85 100644
--- a/src/api/src/router/grpc/ingest/logs.rs
+++ b/src/api/src/router/grpc/ingest/logs.rs
@@ -15,7 +15,7 @@
 
 use config::meta::cluster::get_internal_grpc_token;
 use infra::client::grpc::MetadataMap;
-use openobserve_core::grpc::get_ingester_channel;
+use openobserve_node::grpc::get_ingester_channel;
 use opentelemetry_proto::tonic::collector::logs::v1::{
     ExportLogsServiceRequest, ExportLogsServiceResponse, logs_service_client::LogsServiceClient,
     logs_service_server::LogsService,
diff --git a/src/api/src/router/grpc/ingest/metrics.rs b/src/api/src/router/grpc/ingest/metrics.rs
index 0bad4d1fad..b7be2e3f8a 100644
--- a/src/api/src/router/grpc/ingest/metrics.rs
+++ b/src/api/src/router/grpc/ingest/metrics.rs
@@ -15,7 +15,7 @@
 
 use config::meta::cluster::get_internal_grpc_token;
 use infra::client::grpc::MetadataMap;
-use openobserve_core::grpc::get_ingester_channel;
+use openobserve_node::grpc::get_ingester_channel;
 use opentelemetry_proto::tonic::collector::metrics::v1::{
     ExportMetricsServiceRequest, ExportMetricsServiceResponse,
     metrics_service_client::MetricsServiceClient, metrics_service_server::MetricsService,
diff --git a/src/api/src/router/grpc/ingest/traces.rs b/src/api/src/router/grpc/ingest/traces.rs
index e021a48752..7b9f80f179 100644
--- a/src/api/src/router/grpc/ingest/traces.rs
+++ b/src/api/src/router/grpc/ingest/traces.rs
@@ -15,7 +15,7 @@
 
 use config::meta::cluster::get_internal_grpc_token;
 use infra::client::grpc::MetadataMap;
-use openobserve_core::grpc::get_ingester_channel;
+use openobserve_node::grpc::get_ingester_channel;
 use opentelemetry_proto::tonic::collector::trace::v1::{
     ExportTraceServiceRequest, ExportTraceServiceResponse,
     trace_service_client::TraceServiceClient, trace_service_server::TraceService,
diff --git a/src/api/src/router/http/mod.rs b/src/api/src/router/http/mod.rs
index b242c3a1d0..37d25954ab 100644
--- a/src/api/src/router/http/mod.rs
+++ b/src/api/src/router/http/mod.rs
@@ -49,7 +49,7 @@ static HTTP_CLIENT: OnceLock<reqwest::Client> = OnceLock::new();
 /// Returns a reference to the global HTTP client, initializing it if necessary.
 fn get_http_client() -> &'static reqwest::Client {
     HTTP_CLIENT.get_or_init(|| {
-        openobserve_core::tls::reqwest_client_tls_config()
+        openobserve_tls::reqwest_client_tls_config()
             .expect("Failed to create HTTP client with TLS config")
     })
 }
diff --git a/src/api_management/Cargo.toml b/src/api_management/Cargo.toml
index aaacac8b1b..ca52a6c2b7 100644
--- a/src/api_management/Cargo.toml
+++ b/src/api_management/Cargo.toml
@@ -16,6 +16,7 @@ enterprise = [
     "dep:o2_openfga",
     "openobserve-core/enterprise",
     "openobserve-org-storage/enterprise",
+    "usage_reporting/enterprise",
 ]
 cloud = ["enterprise", "o2_enterprise/cloud", "openobserve-core/cloud"]
 
@@ -29,7 +30,9 @@ search_service.workspace = true
 transform.workspace = true
 vrl.workspace = true
 openobserve-core.workspace = true
+openobserve-node.workspace = true
 openobserve-org-storage.workspace = true
+usage_reporting.workspace = true
 anyhow.workspace = true
 axum.workspace = true
 axum-extra.workspace = true
diff --git a/src/api_management/src/auth/jwt.rs b/src/api_management/src/auth/jwt.rs
index 223745ea74..f76a649565 100644
--- a/src/api_management/src/auth/jwt.rs
+++ b/src/api_management/src/auth/jwt.rs
@@ -826,7 +826,7 @@ async fn publish_org_not_found_error(org_id: &str, user_email: &str) {
         }),
     };
 
-    openobserve_core::self_reporting::publish_error(error_data).await;
+    usage_reporting::publish_error(error_data).await;
 }
 
 #[cfg(feature = "cloud")]
@@ -1157,7 +1157,7 @@ async fn process_custom_claim_parsing(
                     }),
                 };
 
-                openobserve_core::self_reporting::publish_error(error_data).await;
+                usage_reporting::publish_error(error_data).await;
             }) as std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>
         };
 
diff --git a/src/api_management/src/request/organization/org.rs b/src/api_management/src/request/organization/org.rs
index 6bae407aee..2a98b7ffd8 100644
--- a/src/api_management/src/request/organization/org.rs
+++ b/src/api_management/src/request/organization/org.rs
@@ -1431,7 +1431,7 @@ async fn get_super_cluster_nodes(regions: &[String]) -> Result<NodeListResponse,
         let cluster_name = cluster.get_cluster();
 
         // Fetch child nodes from this cluster
-        match openobserve_core::node::get_node_list(&trace_id, cluster).await {
+        match openobserve_node::node::get_node_list(&trace_id, cluster).await {
             Ok(cluster_nodes) => {
                 for node in cluster_nodes {
                     response.add_node(node.clone(), region.clone(), cluster_name.clone());
@@ -1502,7 +1502,7 @@ async fn get_super_cluster_info(regions: &[String]) -> Result<ClusterInfoRespons
         let cluster_name = cluster.get_cluster();
 
         // Fetch cluster info from this cluster node
-        match openobserve_core::cluster_info::get_super_cluster_info(&trace_id, cluster).await {
+        match openobserve_node::cluster_info::get_super_cluster_info(&trace_id, cluster).await {
             Ok(cluster_info_obj) => {
                 response.add_cluster_info(cluster_info_obj, cluster_name.clone(), region.clone());
             }
diff --git a/src/core/Cargo.toml b/src/core/Cargo.toml
index 09c0c55318..10d54a6b91 100644
--- a/src/core/Cargo.toml
+++ b/src/core/Cargo.toml
@@ -46,6 +46,7 @@ vectorscan = [
 db.workspace = true
 audit.workspace = true
 common.workspace = true
+openobserve-node.workspace = true
 transform.workspace = true
 vector-enrichment.workspace = true
 vrl.workspace = true
@@ -93,9 +94,6 @@ proto.workspace = true
 rand.workspace = true
 regex.workspace = true
 reqwest.workspace = true
-rustls.workspace = true
-rustls-native-certs.workspace = true
-webpki-roots.workspace = true
 sea-orm.workspace = true
 serde.workspace = true
 serde_json.workspace = true
@@ -111,7 +109,6 @@ tracing-opentelemetry.workspace = true
 url.workspace = true
 utoipa.workspace = true
 version-compare.workspace = true
-x509-parser.workspace = true
 config.workspace = true
 infra.workspace = true
 ingester.workspace = true
@@ -121,4 +118,3 @@ futures-util.workspace = true
 
 [dev-dependencies]
 multer.workspace = true
-rcgen.workspace = true
diff --git a/src/core/src/alerts/scheduler/handlers.rs b/src/core/src/alerts/scheduler/handlers.rs
index c9822d7716..cdca82d9b6 100644
--- a/src/core/src/alerts/scheduler/handlers.rs
+++ b/src/core/src/alerts/scheduler/handlers.rs
@@ -44,6 +44,7 @@ use infra::{
 #[cfg(feature = "enterprise")]
 use o2_enterprise::enterprise::recommendations::service::QueryRecommendationService;
 use proto::cluster_rpc;
+use usage_reporting::publish_triggers_usage;
 
 #[cfg(feature = "enterprise")]
 use crate::alerts::scheduler::query_optimization_recommendation::QueryOptimizerContext;
@@ -58,7 +59,6 @@ use crate::{
     db::{self, alerts::alert::set_without_updating_trigger},
     ingestion::ingestion_service,
     pipeline::batch_execution::ExecutablePipeline,
-    self_reporting::publish_triggers_usage,
 };
 
 pub async fn handle_triggers(
@@ -161,7 +161,7 @@ async fn handle_anomaly_detection_triggers(
         trigger.status = db::scheduler::TriggerStatus::Waiting;
         db::scheduler::update_trigger(trigger.clone(), true, "").await?;
 
-        crate::self_reporting::publish_triggers_usage(TriggerData {
+        usage_reporting::publish_triggers_usage(TriggerData {
             _timestamp: now_micros(),
             org: trigger.org.clone(),
             module: TriggerDataType::AnomalyDetection,
@@ -221,7 +221,7 @@ async fn handle_anomaly_detection_triggers(
     // Publish trigger run record to the triggers stream (same as alerts).
     let interval_us = parse_detection_interval_to_micros(&config.schedule_interval);
     let next_run = now_micros() + interval_us;
-    crate::self_reporting::publish_triggers_usage(TriggerData {
+    usage_reporting::publish_triggers_usage(TriggerData {
         _timestamp: run_start_us,
         org: trigger.org.clone(),
         module: TriggerDataType::AnomalyDetection,
@@ -2155,7 +2155,7 @@ async fn handle_derived_stream_triggers(
                 error: Some(err_msg),
                 node_errors: HashMap::new(),
             };
-            crate::self_reporting::publish_error(ErrorData {
+            usage_reporting::publish_error(ErrorData {
                 _timestamp: Utc::now().timestamp_micros(),
                 stream_params: pipeline.get_source_stream_params(),
                 error_source: ErrorSource::Pipeline(pipeline_error),
@@ -2303,7 +2303,7 @@ async fn handle_derived_stream_triggers(
                         error: Some(err),
                         node_errors: HashMap::new(),
                     };
-                    crate::self_reporting::publish_error(ErrorData {
+                    usage_reporting::publish_error(ErrorData {
                         _timestamp: Utc::now().timestamp_micros(),
                         stream_params: pipeline.get_source_stream_params(),
                         error_source: ErrorSource::Pipeline(pipeline_error),
@@ -2400,7 +2400,7 @@ async fn handle_derived_stream_triggers(
             error: Some(err_msg),
             node_errors: HashMap::new(),
         };
-        crate::self_reporting::publish_error(ErrorData {
+        usage_reporting::publish_error(ErrorData {
             _timestamp: Utc::now().timestamp_micros(),
             stream_params: pipeline.get_source_stream_params(),
             error_source: ErrorSource::Pipeline(pipeline_error),
diff --git a/src/core/src/bootstrap.rs b/src/core/src/bootstrap.rs
index fdbf716336..bd055503bc 100644
--- a/src/core/src/bootstrap.rs
+++ b/src/core/src/bootstrap.rs
@@ -16,14 +16,45 @@
 use std::sync::Arc;
 
 use ::common::infra::wal;
-use config::{cache_instance_id, ider};
+use config::{cache_instance_id, get_config, ider};
 use db::metas;
 
 #[cfg(feature = "enterprise")]
 use crate::self_reporting::CoreAuditPublisher;
 use crate::self_reporting::persistence::CoreBatchPublisher;
 
+struct CoreOrganizationProvisioner;
+
+#[async_trait::async_trait]
+impl schema::OrganizationProvisioner for CoreOrganizationProvisioner {
+    fn should_auto_create_missing_orgs(&self) -> bool {
+        let cfg = get_config();
+
+        #[cfg(feature = "enterprise")]
+        let usage_enabled = true;
+        #[cfg(not(feature = "enterprise"))]
+        let usage_enabled = cfg.common.usage_enabled;
+
+        #[cfg(feature = "enterprise")]
+        let audit_enabled = o2_enterprise::enterprise::common::config::get_config()
+            .common
+            .audit_enabled;
+        #[cfg(not(feature = "enterprise"))]
+        let audit_enabled = false;
+
+        cfg.common.create_org_through_ingestion || usage_enabled || audit_enabled
+    }
+
+    async fn ensure_org_exists(&self, org_id: &str) -> Result<(), anyhow::Error> {
+        crate::organization::check_and_create_org(org_id)
+            .await
+            .map(|_| ())
+    }
+}
+
 pub async fn init() -> Result<(), anyhow::Error> {
+    schema::set_organization_provisioner(Arc::new(CoreOrganizationProvisioner))
+        .map_err(|_| anyhow::anyhow!("organization provisioner is already initialized"))?;
     usage_reporting::set_batch_publisher(Arc::new(CoreBatchPublisher))
         .map_err(|_| anyhow::anyhow!("usage batch publisher is already initialized"))?;
     #[cfg(feature = "enterprise")]
diff --git a/src/core/src/functions_cache.rs b/src/core/src/functions_cache.rs
index 1343b8ab9a..501f8186dd 100644
--- a/src/core/src/functions_cache.rs
+++ b/src/core/src/functions_cache.rs
@@ -18,8 +18,7 @@ use config::meta::{
     self_reporting::error::{ErrorData, ErrorSource, FunctionError},
     stream::StreamParams,
 };
-
-use crate::self_reporting::publish_error;
+use usage_reporting::publish_error;
 
 /// Loads function records into the shared cache and reports invalid entries.
 pub async fn cache() -> Result<(), anyhow::Error> {
diff --git a/src/core/src/ingestion/ingestion_service.rs b/src/core/src/ingestion/ingestion_service.rs
index faa08c40b8..c943137800 100644
--- a/src/core/src/ingestion/ingestion_service.rs
+++ b/src/core/src/ingestion/ingestion_service.rs
@@ -15,11 +15,10 @@
 
 use config::meta::cluster::get_internal_grpc_token;
 use infra::errors::{Error, Result};
+use openobserve_node::grpc::get_ingester_channel;
 use proto::cluster_rpc;
 use tonic::{Request, codec::CompressionEncoding, metadata::MetadataValue};
 
-use crate::grpc::get_ingester_channel;
-
 pub async fn ingest(req: cluster_rpc::IngestionRequest) -> Result<cluster_rpc::IngestionResponse> {
     let cfg = config::get_config();
     let token: MetadataValue<_> = get_internal_grpc_token()
diff --git a/src/core/src/ingestion/mod.rs b/src/core/src/ingestion/mod.rs
index 5ebdd2ccbe..c54754cf74 100644
--- a/src/core/src/ingestion/mod.rs
+++ b/src/core/src/ingestion/mod.rs
@@ -44,12 +44,12 @@ use infra::{
 };
 use ingestion_common::IngestionRequest;
 use proto::cluster_rpc::IngestionType;
+use usage_reporting::publish_triggers_usage;
 use vrl::compiler::runtime::Runtime;
 
 use super::{
     db::alerts::alert,
     pipeline::{batch_execution::ExecutablePipeline, db as pipeline},
-    self_reporting::publish_triggers_usage,
 };
 use crate::{
     alerts::alert::AlertExt,
diff --git a/src/core/src/lib.rs b/src/core/src/lib.rs
index 35d4d4bc50..8c340c8c0d 100644
--- a/src/core/src/lib.rs
+++ b/src/core/src/lib.rs
@@ -24,7 +24,6 @@ pub mod auth;
 pub mod authz;
 pub mod bootstrap;
 pub mod cache;
-pub mod cluster_info;
 pub mod dashboards;
 use ::common;
 use ::db;
@@ -32,7 +31,6 @@ pub mod error_suggest;
 use ::db::folders;
 pub mod functions;
 pub mod functions_cache;
-pub mod grpc;
 pub mod http;
 pub mod ingestion;
 pub mod ingestion_tokens;
@@ -42,7 +40,6 @@ pub mod llm_evaluations;
 pub mod logs;
 pub mod metadata;
 pub mod metrics;
-pub mod node;
 #[cfg(feature = "enterprise")]
 pub mod ofga;
 pub mod org_cleanup;
@@ -54,8 +51,6 @@ pub mod pipeline;
 pub mod providers;
 #[cfg(feature = "enterprise")]
 pub mod ratelimit;
-pub mod runtime_metrics;
-pub mod schema_watcher;
 use search_service as search;
 pub mod self_reporting;
 pub mod service;
@@ -65,7 +60,6 @@ pub mod stream;
 pub mod stream_utils;
 pub mod synthetics;
 pub mod system_settings;
-pub mod tls;
 pub mod traces;
 #[cfg(feature = "cloud")]
 pub mod trial_quota;
diff --git a/src/core/src/llm_evaluations/evaluator_trace_exporter.rs b/src/core/src/llm_evaluations/evaluator_trace_exporter.rs
index 1fc58b8c1f..8fc2991002 100644
--- a/src/core/src/llm_evaluations/evaluator_trace_exporter.rs
+++ b/src/core/src/llm_evaluations/evaluator_trace_exporter.rs
@@ -44,7 +44,7 @@ impl EvaluatorTraceExporter {
             }
         };
 
-        let (_addr, channel) = match crate::grpc::get_ingester_channel().await {
+        let (_addr, channel) = match openobserve_node::grpc::get_ingester_channel().await {
             Ok(v) => v,
             Err(e) => {
                 log::error!(
diff --git a/src/core/src/pipeline/batch_execution.rs b/src/core/src/pipeline/batch_execution.rs
index 91bfe5fb45..4e8450df71 100644
--- a/src/core/src/pipeline/batch_execution.rs
+++ b/src/core/src/pipeline/batch_execution.rs
@@ -53,11 +53,9 @@ use transform::{
     QUERY_FUNCTIONS, apply_vrl_fn, compile_vrl_function,
     js::{JSRuntimeConfig, apply_js_fn, compile_js_function},
 };
+use usage_reporting::publish_error;
 
-use crate::{
-    alerts::{ConditionExt, ConditionGroupExt},
-    self_reporting::publish_error,
-};
+use crate::alerts::{ConditionExt, ConditionGroupExt};
 
 // Global batch buffer for accumulating remote stream records
 #[cfg(feature = "enterprise")]
diff --git a/src/core/src/self_reporting/mod.rs b/src/core/src/self_reporting/mod.rs
index 369fecca02..ca0b84d20b 100644
--- a/src/core/src/self_reporting/mod.rs
+++ b/src/core/src/self_reporting/mod.rs
@@ -13,17 +13,6 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-use std::time::Duration;
-
-#[cfg(feature = "enterprise")]
-use audit::flush as flush_audit;
-use config::{
-    cluster::LOCAL_NODE,
-    get_config,
-    meta::self_reporting::{EnqueueError, ReportingData, error::ErrorData, usage::TriggerData},
-    metrics,
-};
-
 #[cfg(feature = "cloud")]
 pub mod cloud_events;
 #[cfg(feature = "enterprise")]
@@ -39,178 +28,6 @@ mod usage_schema;
 #[cfg(feature = "cloud")]
 pub use ingestion::ingest_data_retention_usages;
 
-pub async fn run() {
-    #[cfg(not(feature = "enterprise"))]
-    {
-        let cfg = get_config();
-        if !cfg.common.usage_enabled {
-            return;
-        }
-    }
-
-    if let Err(error) = usage_reporting::start().await {
-        log::error!("[SELF-REPORTING] Reporting queue initialization failed: {error}");
-        return;
-    }
-
-    log::debug!("[SELF-REPORTING] successfully initialized reporting queues");
-}
-
-pub fn publish_triggers_usage(trigger: TriggerData) {
-    #[cfg(not(feature = "enterprise"))]
-    {
-        let cfg = get_config();
-        if !cfg.common.usage_enabled {
-            log::debug!(
-                "[SELF-REPORTING] Skipping trigger usage publish - usage reporting disabled"
-            );
-            return;
-        }
-    }
-
-    log::debug!(
-        "[SELF-REPORTING] Publishing trigger usage: org={}, module={:?}, key={}, status={:?}",
-        trigger.org,
-        trigger.module,
-        trigger.key,
-        trigger.status
-    );
-
-    // Use non-blocking try_enqueue to prevent scheduler blocking
-    match usage_reporting::try_enqueue(ReportingData::Trigger(Box::new(trigger))) {
-        Ok(()) => {
-            log::debug!(
-                "[SELF-REPORTING] Successfully queued trigger usage data to be ingested (non-blocking)"
-            )
-        }
-        Err(tokio::sync::mpsc::error::TrySendError::Full(msg)) => {
-            // Queue is full - system is overloaded, drop the trigger data
-            let dropped_info = match msg {
-                config::meta::self_reporting::ReportingMessage::Data(ReportingData::Trigger(t)) => {
-                    Some((t.org.clone(), t.key.clone()))
-                }
-                _ => None,
-            };
-
-            if let Some((org, key)) = dropped_info {
-                log::warn!(
-                    "[SELF-REPORTING] Usage queue full, dropping trigger data for org={}, key={}. \
-                     System is overloaded. Consider increasing ZO_USAGE_REPORTING_THREAD_NUM or ZO_USAGE_BATCH_SIZE.",
-                    org,
-                    key
-                );
-            } else {
-                log::warn!(
-                    "[SELF-REPORTING] Usage queue full, dropping trigger data. \
-                     System is overloaded. Consider increasing ZO_USAGE_REPORTING_THREAD_NUM or ZO_USAGE_BATCH_SIZE."
-                );
-            }
-
-            // Increment dropped triggers metric
-            metrics::SELF_REPORTING_DROPPED_TRIGGERS
-                .with_label_values(&["usage"])
-                .inc();
-        }
-        Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
-            log::error!(
-                "[SELF-REPORTING] Usage queue closed, cannot send trigger data. \
-                 Self-reporting service may have shut down."
-            );
-        }
-    }
-}
-
-pub async fn publish_error(error_data: ErrorData) {
-    let cfg = get_config();
-    #[cfg(not(feature = "enterprise"))]
-    {
-        if !cfg.common.usage_enabled {
-            log::debug!("[SELF-REPORTING] Skipping error publish - usage reporting disabled");
-            return;
-        }
-    }
-
-    // Important data - use shorter timeout than usage to prevent indefinite blocking
-    let timeout_duration = Duration::from_secs(cfg.common.error_publish_timeout_secs);
-
-    // Extract org for logging
-    let org = error_data.stream_params.org_id.clone();
-    let error_source = match &error_data.error_source {
-        config::meta::self_reporting::error::ErrorSource::Alert => "Alert",
-        config::meta::self_reporting::error::ErrorSource::Dashboard => "Dashboard",
-        config::meta::self_reporting::error::ErrorSource::Function(_) => "Function",
-        config::meta::self_reporting::error::ErrorSource::Ingestion => "Ingestion",
-        config::meta::self_reporting::error::ErrorSource::Pipeline(_) => "Pipeline",
-        config::meta::self_reporting::error::ErrorSource::Search => "Search",
-        config::meta::self_reporting::error::ErrorSource::Other => "Other",
-        config::meta::self_reporting::error::ErrorSource::SsoClaimParser(_) => "SsoClaimParser",
-        config::meta::self_reporting::error::ErrorSource::OrgStorage(_) => "OrgStorage",
-    };
-
-    log::debug!(
-        "[SELF-REPORTING] Publishing error data: org={}, source={}, timeout={:?}",
-        org,
-        error_source,
-        timeout_duration
-    );
-
-    let start = std::time::Instant::now();
-    match usage_reporting::enqueue_error_with_timeout(
-        ReportingData::Error(Box::new(error_data)),
-        timeout_duration,
-    )
-    .await
-    {
-        Ok(()) => {
-            let elapsed = start.elapsed();
-            log::debug!(
-                "[SELF-REPORTING] Successfully queued error data to be ingested (took {:?}, timeout-based)",
-                elapsed
-            );
-        }
-        Err(EnqueueError::Timeout) => {
-            log::warn!(
-                "[SELF-REPORTING] Timeout ({:?}) queueing error data for org={}, source={}. \
-                 System overloaded, error reporting degraded. \
-                 Consider increasing ZO_USAGE_REPORTING_THREAD_NUM or ZO_USAGE_BATCH_SIZE.",
-                timeout_duration,
-                org,
-                error_source
-            );
-            metrics::SELF_REPORTING_TIMEOUT_ERRORS
-                .with_label_values(&["error"])
-                .inc();
-        }
-        Err(EnqueueError::SendFailed(e)) => {
-            log::error!(
-                "[SELF-REPORTING] Failed to send error data for org={}, source={}: {e}",
-                org,
-                error_source
-            );
-        }
-    }
-}
-
-pub async fn flush() {
-    // flush audit data
-    #[cfg(feature = "enterprise")]
-    flush_audit().await;
-
-    let cfg = get_config();
-
-    #[cfg(feature = "enterprise")]
-    let usage_enabled = true;
-    #[cfg(not(feature = "enterprise"))]
-    let usage_enabled = cfg.common.usage_enabled;
-
-    // only ingester and querier nodes report usage
-    if !usage_enabled || (!LOCAL_NODE.is_ingester() && !LOCAL_NODE.is_querier()) {
-        return;
-    }
-
-    usage_reporting::shutdown(cfg.limit.usage_reporting_thread_num).await;
-}
-
 #[cfg(feature = "enterprise")]
 pub struct CoreAuditPublisher;
 
diff --git a/src/core/src/workflows/mod.rs b/src/core/src/workflows/mod.rs
index f22faa2665..3fc27dd638 100644
--- a/src/core/src/workflows/mod.rs
+++ b/src/core/src/workflows/mod.rs
@@ -28,11 +28,11 @@ use infra::table::workflows::{self, Workflow, WorkflowError, WorkflowRunData, Wo
 use o2_enterprise::enterprise::common::config::get_config as get_o2_config;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
+use usage_reporting::publish_triggers_usage;
 
 use crate::{
     common::{infra::config::ALERTS, meta::authz::Authz, utils::get_nats_lock},
     pipeline::batch_execution::{ExecutablePipeline, WorkflowResult},
-    self_reporting::publish_triggers_usage,
 };
 
 pub mod runtime;
diff --git a/src/jobs/Cargo.toml b/src/jobs/Cargo.toml
index e409231e2c..da028e26a4 100644
--- a/src/jobs/Cargo.toml
+++ b/src/jobs/Cargo.toml
@@ -22,6 +22,7 @@ enterprise = [
     "search_service/enterprise",
     "stream/enterprise",
     "tantivy_utils/enterprise",
+    "usage_reporting/enterprise",
 ]
 cloud = [
     "enterprise",
@@ -41,6 +42,7 @@ audit.workspace = true
 common.workspace = true
 db.workspace = true
 openobserve-core.workspace = true
+openobserve-node.workspace = true
 openobserve-builtins.workspace = true
 openobserve-org-storage.workspace = true
 ingestion-common.workspace = true
@@ -52,6 +54,7 @@ promql-service.workspace = true
 search.workspace = true
 search_service.workspace = true
 tantivy_utils.workspace = true
+usage_reporting.workspace = true
 anyhow.workspace = true
 async-trait.workspace = true
 arrow.workspace = true
diff --git a/src/jobs/src/job/mod.rs b/src/jobs/src/job/mod.rs
index 5cdf666d4b..655722c3e1 100644
--- a/src/jobs/src/job/mod.rs
+++ b/src/jobs/src/job/mod.rs
@@ -15,7 +15,9 @@
 
 use config::{DEFAULT_ORG, cluster::LOCAL_NODE, spawn_pausable_job};
 use db;
-use openobserve_core::{alerts, self_reporting, users};
+#[cfg(feature = "cloud")]
+use openobserve_core::self_reporting;
+use openobserve_core::{alerts, users};
 use regex::Regex;
 #[cfg(feature = "enterprise")]
 use {
@@ -489,7 +491,7 @@ pub async fn init() -> Result<(), anyhow::Error> {
         );
     }
 
-    tokio::task::spawn(self_reporting::run());
+    tokio::task::spawn(usage_reporting::run());
 
     // cache short_urls
     tokio::task::spawn(db::short_url::watch());
@@ -498,7 +500,8 @@ pub async fn init() -> Result<(), anyhow::Error> {
         .expect("short url cache failed");
 
     // initialize metadata watcher
-    tokio::task::spawn(openobserve_core::schema_watcher::watch());
+    let schema_watcher = schema::create_watcher()?;
+    tokio::task::spawn(schema_watcher);
     tokio::task::spawn(db::functions::watch());
     tokio::task::spawn(db::compact::retention::watch());
     tokio::task::spawn(db::metrics::watch_prom_cluster_leader());
@@ -903,7 +906,7 @@ pub async fn init() -> Result<(), anyhow::Error> {
                     use config::meta::self_reporting::usage::{
                         TriggerData, TriggerDataStatus, TriggerDataType,
                     };
-                    openobserve_core::self_reporting::publish_triggers_usage(TriggerData {
+                    usage_reporting::publish_triggers_usage(TriggerData {
                         _timestamp: start_us,
                         org: org_id,
                         module: TriggerDataType::AnomalyDetectionTraining,
diff --git a/src/jobs/src/job/org_storage.rs b/src/jobs/src/job/org_storage.rs
index 605170550b..0aa9687c19 100644
--- a/src/jobs/src/job/org_storage.rs
+++ b/src/jobs/src/job/org_storage.rs
@@ -82,7 +82,7 @@ async fn test_storage_validity() {
                     "org {} has been removed from org storage cache, and will use default storage",
                     provider.org_id
                 );
-                openobserve_core::self_reporting::publish_error(ErrorData {
+                usage_reporting::publish_error(ErrorData {
                     _timestamp: chrono::Utc::now().timestamp_micros(),
                     stream_params: Default::default(),
                     error_source: ErrorSource::OrgStorage(OrgStorageError {
diff --git a/src/jobs/src/job/promql_self_consume.rs b/src/jobs/src/job/promql_self_consume.rs
index 4efdf0dbfc..5a8a6057e9 100644
--- a/src/jobs/src/job/promql_self_consume.rs
+++ b/src/jobs/src/job/promql_self_consume.rs
@@ -24,6 +24,7 @@ use config::{
 };
 use hashbrown::HashSet;
 use ingestion_common::{IngestUser, SystemJobType};
+use openobserve_node::grpc::get_ingester_channel;
 use proto::cluster_rpc::{
     IngestionData, IngestionRequest, IngestionType, ingest_client::IngestClient,
 };
@@ -35,7 +36,7 @@ use tonic::{
     metadata::{MetadataKey, MetadataValue},
 };
 
-use crate::service::{self, grpc::get_ingester_channel};
+use crate::service;
 
 static METRICS_WHITELIST: Lazy<HashSet<String>> = Lazy::new(|| {
     config::get_config()
diff --git a/src/main.rs b/src/main.rs
index b91da13048..67844e6fc9 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -56,10 +56,9 @@ use openobserve_api::{
     },
     router,
 };
-use openobserve_core::{
-    bootstrap, cluster_info::ClusterInfoService, metadata, node::NodeService, self_reporting,
-};
+use openobserve_core::{bootstrap, metadata};
 use openobserve_jobs::job;
+use openobserve_node::{cluster_info::ClusterInfoService, node::NodeService};
 use opentelemetry::{KeyValue, global, trace::TracerProvider};
 use opentelemetry_otlp::{WithExportConfig, WithHttpConfig, WithTonicConfig};
 use opentelemetry_proto::tonic::collector::{
@@ -108,6 +107,13 @@ use tracing_subscriber::{
 #[unsafe(export_name = "malloc_conf")]
 pub static malloc_conf: &[u8] = b"prof:true,prof_active:true,lg_prof_sample:16\0";
 
+async fn flush_reporting() {
+    #[cfg(feature = "enterprise")]
+    audit::flush().await;
+
+    usage_reporting::flush().await;
+}
+
 #[tokio::main]
 async fn main() -> Result<(), anyhow::Error> {
     // CLI provides the path to the config file (if any)
@@ -269,7 +275,7 @@ async fn main() -> Result<(), anyhow::Error> {
 
             // Register job runtime for metrics collection
             if let Ok(handle) = tokio::runtime::Handle::try_current() {
-                openobserve_core::runtime_metrics::register_runtime("job".to_string(), handle);
+                openobserve_node::runtime_metrics::register_runtime("job".to_string(), handle);
             }
 
             job_init_tx.send(true).ok();
@@ -310,7 +316,7 @@ async fn main() -> Result<(), anyhow::Error> {
         };
 
         // Register gRPC runtime for metrics collection
-        openobserve_core::runtime_metrics::register_runtime(
+        openobserve_node::runtime_metrics::register_runtime(
             "grpc".to_string(),
             rt.handle().clone(),
         );
@@ -334,11 +340,11 @@ async fn main() -> Result<(), anyhow::Error> {
 
     // Register main HTTP runtime for metrics collection
     if let Ok(handle) = tokio::runtime::Handle::try_current() {
-        openobserve_core::runtime_metrics::register_runtime("http".to_string(), handle);
+        openobserve_node::runtime_metrics::register_runtime("http".to_string(), handle);
     }
 
     // Start runtime metrics collector
-    openobserve_core::runtime_metrics::start_metrics_collector().await;
+    openobserve_node::runtime_metrics::start_metrics_collector().await;
 
     // let node online
     let _ = cluster::set_online().await;
@@ -429,7 +435,7 @@ async fn main() -> Result<(), anyhow::Error> {
     }
 
     // flush usage report
-    self_reporting::flush().await;
+    flush_reporting().await;
 
     // flush service discovery
     #[cfg(feature = "enterprise")]
@@ -990,7 +996,7 @@ impl opentelemetry_sdk::trace::SpanExporter for MetaOrgTraceExporter {
                     }
                 };
 
-                let (_addr, channel) = match openobserve_core::grpc::get_ingester_channel().await {
+                let (_addr, channel) = match openobserve_node::grpc::get_ingester_channel().await {
                     Ok(v) => v,
                     Err(e) => {
                         log::error!("[SEARCH-INSPECTOR] Failed to get ingester channel: {e}");
@@ -1350,7 +1356,7 @@ async fn init_action_server() -> Result<(), anyhow::Error> {
     log::info!("HTTP server stopped");
 
     // flush usage report
-    self_reporting::flush().await;
+    flush_reporting().await;
 
     // stop telemetry
     if cfg.common.telemetry_enabled {
diff --git a/src/node/Cargo.toml b/src/node/Cargo.toml
new file mode 100644
index 0000000000..6d4f49fe9f
--- /dev/null
+++ b/src/node/Cargo.toml
@@ -0,0 +1,21 @@
+[package]
+name = "openobserve-node"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+publish = false
+
+[lib]
+path = "src/lib.rs"
+
+[dependencies]
+anyhow.workspace = true
+common.workspace = true
+config.workspace = true
+db.workspace = true
+infra.workspace = true
+log.workspace = true
+prometheus.workspace = true
+proto.workspace = true
+tokio.workspace = true
+tonic.workspace = true
diff --git a/src/node/build.rs b/src/node/build.rs
new file mode 100644
index 0000000000..256b344dee
--- /dev/null
+++ b/src/node/build.rs
@@ -0,0 +1,18 @@
+// Copyright 2026 OpenObserve Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+fn main() {
+    println!("cargo:rustc-check-cfg=cfg(tokio_unstable)");
+}
diff --git a/src/core/src/cluster_info.rs b/src/node/src/cluster_info.rs
similarity index 98%
rename from src/core/src/cluster_info.rs
rename to src/node/src/cluster_info.rs
index 5b778c86c1..2b1d439fcf 100644
--- a/src/core/src/cluster_info.rs
+++ b/src/node/src/cluster_info.rs
@@ -69,7 +69,7 @@ impl proto::cluster_rpc::cluster_info_service_server::ClusterInfoService for Clu
         let req = request.into_inner();
 
         // Get the key from the database
-        match crate::db::compact::compactor_manual_jobs::get_job(&req.ksuid).await {
+        match db::compact::compactor_manual_jobs::get_job(&req.ksuid).await {
             Ok(job) => {
                 // Job still exists, return pending status
                 Ok(Response::new(GetDeleteJobStatusResponse {
diff --git a/src/core/src/grpc.rs b/src/node/src/grpc.rs
similarity index 100%
rename from src/core/src/grpc.rs
rename to src/node/src/grpc.rs
diff --git a/src/node/src/lib.rs b/src/node/src/lib.rs
new file mode 100644
index 0000000000..15c018b929
--- /dev/null
+++ b/src/node/src/lib.rs
@@ -0,0 +1,19 @@
+// Copyright 2026 OpenObserve Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+pub mod cluster_info;
+pub mod grpc;
+pub mod node;
+pub mod runtime_metrics;
diff --git a/src/core/src/node.rs b/src/node/src/node.rs
similarity index 100%
rename from src/core/src/node.rs
rename to src/node/src/node.rs
diff --git a/src/core/src/runtime_metrics.rs b/src/node/src/runtime_metrics.rs
similarity index 100%
rename from src/core/src/runtime_metrics.rs
rename to src/node/src/runtime_metrics.rs
diff --git a/src/schema/Cargo.toml b/src/schema/Cargo.toml
index 6784b582da..8a892646e4 100644
--- a/src/schema/Cargo.toml
+++ b/src/schema/Cargo.toml
@@ -15,15 +15,16 @@ enterprise = ["config/enterprise", "db/enterprise"]
 [dependencies]
 anyhow.workspace = true
 arrow-schema.workspace = true
+async-trait.workspace = true
 common.workspace = true
 config.workspace = true
 dashmap.workspace = true
 db.workspace = true
 hashbrown.workspace = true
 infra.workspace = true
+ingester.workspace = true
+itertools.workspace = true
 log.workspace = true
 ingestion-common.workspace = true
 serde_json.workspace = true
-
-[dev-dependencies]
 tokio.workspace = true
diff --git a/src/schema/src/lib.rs b/src/schema/src/lib.rs
index a99711aa5b..8ef3bedd9a 100644
--- a/src/schema/src/lib.rs
+++ b/src/schema/src/lib.rs
@@ -53,6 +53,12 @@ use infra::schema::{
 use ingestion_common::StreamSchemaChk;
 use serde_json::{Map, Value};
 
+mod watcher;
+
+pub use watcher::{
+    OrganizationProvisioner, SchemaWatcher, create_watcher, set_organization_provisioner,
+};
+
 const SCHEMA_CONFORMANCE_FAILED: &str = "schema_conformance_failed";
 
 pub fn get_upto_discard_error() -> anyhow::Error {
diff --git a/src/core/src/schema_watcher.rs b/src/schema/src/watcher.rs
similarity index 90%
rename from src/core/src/schema_watcher.rs
rename to src/schema/src/watcher.rs
index 71a89f3b55..87fde3099e 100644
--- a/src/core/src/schema_watcher.rs
+++ b/src/schema/src/watcher.rs
@@ -13,14 +13,18 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-//! Runtime watcher for schema cache updates.
+//! Runtime watcher for schema cache updates and stream-local cache cleanup.
 
 use std::{
-    sync::{Arc, atomic::Ordering},
+    future::Future,
+    pin::Pin,
+    sync::{Arc, OnceLock, atomic::Ordering},
     time::Duration,
 };
 
 use arrow_schema::Schema;
+use async_trait::async_trait;
+use common::infra::config::ORGANIZATIONS;
 use config::{
     cluster::LOCAL_NODE_ID,
     get_config,
@@ -29,7 +33,6 @@ use config::{
     meta::stream::StreamType,
     utils::{json, time::now_micros},
 };
-use db;
 use infra::{
     cache,
     schema::{
@@ -37,17 +40,35 @@ use infra::{
         SchemaCache, unwrap_stream_settings,
     },
 };
-#[cfg(feature = "enterprise")]
-use o2_enterprise::enterprise::common::config::get_config as get_o2_config;
 
-use crate::{common::infra::config::ORGANIZATIONS, organization::check_and_create_org};
+#[async_trait]
+pub trait OrganizationProvisioner: Send + Sync + 'static {
+    fn should_auto_create_missing_orgs(&self) -> bool;
 
-pub async fn watch() -> Result<(), anyhow::Error> {
-    #[cfg(feature = "enterprise")]
-    let audit_enabled = get_o2_config().common.audit_enabled;
-    #[cfg(not(feature = "enterprise"))]
-    let audit_enabled = false;
-    let cfg = get_config();
+    async fn ensure_org_exists(&self, org_id: &str) -> Result<(), anyhow::Error>;
+}
+
+static ORGANIZATION_PROVISIONER: OnceLock<Arc<dyn OrganizationProvisioner>> = OnceLock::new();
+
+pub type SchemaWatcher = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + 'static>>;
+
+pub fn set_organization_provisioner(
+    provisioner: Arc<dyn OrganizationProvisioner>,
+) -> Result<(), Arc<dyn OrganizationProvisioner>> {
+    ORGANIZATION_PROVISIONER.set(provisioner)
+}
+
+pub fn create_watcher() -> Result<SchemaWatcher, anyhow::Error> {
+    let organization_provisioner = ORGANIZATION_PROVISIONER
+        .get()
+        .cloned()
+        .ok_or_else(|| anyhow::anyhow!("organization provisioner is not initialized"))?;
+    Ok(Box::pin(watch(organization_provisioner)))
+}
+
+async fn watch(
+    organization_provisioner: Arc<dyn OrganizationProvisioner>,
+) -> Result<(), anyhow::Error> {
     let key = "/schema/";
     let cluster_coordinator = infra::db::get_coordinator().await;
     let mut events = cluster_coordinator.watch(key).await?;
@@ -175,16 +196,11 @@ pub async fn watch() -> Result<(), anyhow::Error> {
                 let keys = item_key.split('/').collect::<Vec<&str>>();
                 let org_id = keys[0];
 
-                #[cfg(feature = "enterprise")]
-                let usage_enabled = true;
-                #[cfg(not(feature = "enterprise"))]
-                let usage_enabled = cfg.common.usage_enabled;
-
                 // if create_org_through_ingestion is enabled, we need to create the org
                 // if it doesn't exist. Hence, we need to check if the org exists in the cache
-                if (cfg.common.create_org_through_ingestion || usage_enabled || audit_enabled)
+                if organization_provisioner.should_auto_create_missing_orgs()
                     && !ORGANIZATIONS.read().await.contains_key(org_id)
-                    && let Err(e) = check_and_create_org(org_id).await
+                    && let Err(e) = organization_provisioner.ensure_org_exists(org_id).await
                 {
                     log::error!("Failed to save organization in database: {e}");
                 }
diff --git a/src/tls/Cargo.toml b/src/tls/Cargo.toml
new file mode 100644
index 0000000000..a8c3fa9651
--- /dev/null
+++ b/src/tls/Cargo.toml
@@ -0,0 +1,23 @@
+[package]
+name = "openobserve-tls"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+publish = false
+
+[lib]
+path = "src/lib.rs"
+
+[dependencies]
+anyhow.workspace = true
+config.workspace = true
+itertools.workspace = true
+log.workspace = true
+reqwest.workspace = true
+rustls.workspace = true
+rustls-native-certs.workspace = true
+webpki-roots.workspace = true
+x509-parser.workspace = true
+
+[dev-dependencies]
+rcgen.workspace = true
diff --git a/src/core/src/tls/mod.rs b/src/tls/src/lib.rs
similarity index 99%
rename from src/core/src/tls/mod.rs
rename to src/tls/src/lib.rs
index 7a139b4b25..77977a8a39 100644
--- a/src/core/src/tls/mod.rs
+++ b/src/tls/src/lib.rs
@@ -13,6 +13,8 @@
 // You should have received a copy of the GNU Affero General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+//! TLS configuration and certificate utilities.
+
 use std::{
     io::BufReader,
     net::{Ipv4Addr, Ipv6Addr},
diff --git a/src/usage_reporting/src/lib.rs b/src/usage_reporting/src/lib.rs
index a20053588f..aead0fd333 100644
--- a/src/usage_reporting/src/lib.rs
+++ b/src/usage_reporting/src/lib.rs
@@ -5,16 +5,22 @@
 // the Free Software Foundation, either version 3 of the License, or
 // (at your option) any later version.
 
-use std::sync::{Arc, LazyLock as Lazy, OnceLock};
+use std::{
+    sync::{Arc, LazyLock as Lazy, OnceLock},
+    time::Duration,
+};
 
 use async_trait::async_trait;
 use chrono::{DateTime, Datelike, Timelike};
 use config::{
-    SIZE_IN_MB, get_config,
+    SIZE_IN_MB,
+    cluster::LOCAL_NODE,
+    get_config,
     meta::{
         self_reporting::{
-            ReportingData, ReportingMessage, ReportingQueue, ReportingRunner,
-            usage::{RequestStats, UsageData, UsageEvent, UsageType},
+            EnqueueError, ReportingData, ReportingMessage, ReportingQueue, ReportingRunner,
+            error::ErrorData,
+            usage::{RequestStats, TriggerData, UsageData, UsageEvent, UsageType},
         },
         stream::StreamType,
     },
@@ -43,6 +49,161 @@ pub fn set_batch_publisher(
     BATCH_PUBLISHER.set(publisher)
 }
 
+pub async fn run() {
+    #[cfg(not(feature = "enterprise"))]
+    {
+        if !get_config().common.usage_enabled {
+            return;
+        }
+    }
+
+    if let Err(error) = start().await {
+        log::error!("[SELF-REPORTING] Reporting queue initialization failed: {error}");
+        return;
+    }
+
+    log::debug!("[SELF-REPORTING] successfully initialized reporting queues");
+}
+
+pub fn publish_triggers_usage(trigger: TriggerData) {
+    #[cfg(not(feature = "enterprise"))]
+    {
+        if !get_config().common.usage_enabled {
+            log::debug!(
+                "[SELF-REPORTING] Skipping trigger usage publish - usage reporting disabled"
+            );
+            return;
+        }
+    }
+
+    log::debug!(
+        "[SELF-REPORTING] Publishing trigger usage: org={}, module={:?}, key={}, status={:?}",
+        trigger.org,
+        trigger.module,
+        trigger.key,
+        trigger.status
+    );
+
+    match try_enqueue(ReportingData::Trigger(Box::new(trigger))) {
+        Ok(()) => {
+            log::debug!(
+                "[SELF-REPORTING] Successfully queued trigger usage data to be ingested (non-blocking)"
+            )
+        }
+        Err(tokio::sync::mpsc::error::TrySendError::Full(message)) => {
+            let dropped_info = match message {
+                ReportingMessage::Data(ReportingData::Trigger(trigger)) => {
+                    Some((trigger.org.clone(), trigger.key.clone()))
+                }
+                _ => None,
+            };
+
+            if let Some((org, key)) = dropped_info {
+                log::warn!(
+                    "[SELF-REPORTING] Usage queue full, dropping trigger data for org={}, key={}. \
+                     System is overloaded. Consider increasing ZO_USAGE_REPORTING_THREAD_NUM or ZO_USAGE_BATCH_SIZE.",
+                    org,
+                    key
+                );
+            } else {
+                log::warn!(
+                    "[SELF-REPORTING] Usage queue full, dropping trigger data. \
+                     System is overloaded. Consider increasing ZO_USAGE_REPORTING_THREAD_NUM or ZO_USAGE_BATCH_SIZE."
+                );
+            }
+
+            metrics::SELF_REPORTING_DROPPED_TRIGGERS
+                .with_label_values(&["usage"])
+                .inc();
+        }
+        Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
+            log::error!(
+                "[SELF-REPORTING] Usage queue closed, cannot send trigger data. \
+                 Self-reporting service may have shut down."
+            );
+        }
+    }
+}
+
+pub async fn publish_error(error_data: ErrorData) {
+    let cfg = get_config();
+    #[cfg(not(feature = "enterprise"))]
+    {
+        if !cfg.common.usage_enabled {
+            log::debug!("[SELF-REPORTING] Skipping error publish - usage reporting disabled");
+            return;
+        }
+    }
+
+    let timeout_duration = Duration::from_secs(cfg.common.error_publish_timeout_secs);
+    let org = error_data.stream_params.org_id.clone();
+    let error_source = match &error_data.error_source {
+        config::meta::self_reporting::error::ErrorSource::Alert => "Alert",
+        config::meta::self_reporting::error::ErrorSource::Dashboard => "Dashboard",
+        config::meta::self_reporting::error::ErrorSource::Function(_) => "Function",
+        config::meta::self_reporting::error::ErrorSource::Ingestion => "Ingestion",
+        config::meta::self_reporting::error::ErrorSource::Pipeline(_) => "Pipeline",
+        config::meta::self_reporting::error::ErrorSource::Search => "Search",
+        config::meta::self_reporting::error::ErrorSource::Other => "Other",
+        config::meta::self_reporting::error::ErrorSource::SsoClaimParser(_) => "SsoClaimParser",
+        config::meta::self_reporting::error::ErrorSource::OrgStorage(_) => "OrgStorage",
+    };
+
+    log::debug!(
+        "[SELF-REPORTING] Publishing error data: org={}, source={}, timeout={:?}",
+        org,
+        error_source,
+        timeout_duration
+    );
+
+    let start = std::time::Instant::now();
+    match enqueue_error_with_timeout(ReportingData::Error(Box::new(error_data)), timeout_duration)
+        .await
+    {
+        Ok(()) => {
+            log::debug!(
+                "[SELF-REPORTING] Successfully queued error data to be ingested (took {:?}, timeout-based)",
+                start.elapsed()
+            );
+        }
+        Err(EnqueueError::Timeout) => {
+            log::warn!(
+                "[SELF-REPORTING] Timeout ({:?}) queueing error data for org={}, source={}. \
+                 System overloaded, error reporting degraded. \
+                 Consider increasing ZO_USAGE_REPORTING_THREAD_NUM or ZO_USAGE_BATCH_SIZE.",
+                timeout_duration,
+                org,
+                error_source
+            );
+            metrics::SELF_REPORTING_TIMEOUT_ERRORS
+                .with_label_values(&["error"])
+                .inc();
+        }
+        Err(EnqueueError::SendFailed(error)) => {
+            log::error!(
+                "[SELF-REPORTING] Failed to send error data for org={}, source={}: {error}",
+                org,
+                error_source
+            );
+        }
+    }
+}
+
+pub async fn flush() {
+    let cfg = get_config();
+
+    #[cfg(feature = "enterprise")]
+    let usage_enabled = true;
+    #[cfg(not(feature = "enterprise"))]
+    let usage_enabled = cfg.common.usage_enabled;
+
+    if !usage_enabled || (!LOCAL_NODE.is_ingester() && !LOCAL_NODE.is_querier()) {
+        return;
+    }
+
+    shutdown(cfg.limit.usage_reporting_thread_num).await;
+}
+
 pub async fn report_request_usage_stats(
     stats: RequestStats,
     org_id: &str,