use std::collections::BTreeSet;
use std::fs;
use std::io::Cursor;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use base64::Engine as _;
use image::{DynamicImage, GenericImageView, ImageFormat, imageops::FilterType};
use rand::RngExt;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use crate::platform::adapters::data::DataAdapter;
use crate::platform::error::PlatformError;
use crate::platform::model::{
CreateHubTokenRequest, CreateProjectRequest, HubAccessGrant, HubAssetPackage, HubAssetVersion,
HubAuthority, HubPublisher, HubToken, PlatformHubRepository, PlatformServiceInstance,
ProjectFileLayout, ProjectHubRepository, ProjectRuntimeSelectionRequest, ZebflowJson, now_ts,
slug_segment,
};
use crate::platform::policy::package::{
PackagePolicyEntry, PackageReviewOptions, PackageSafetyReview, review_package_entries,
};
use crate::platform::sekejap;
use crate::platform::services::ProjectService;
use crate::platform::services::project::derive_trigger_kind_from_source;
use crate::platform::services::tsx_outline::extract_import_sources;
use crate::platform::sqlite_schema;
use crate::zebfs::{LocalZebFs, normalize_object_path};
pub struct HubService {
control_data: Arc<dyn DataAdapter>,
hub_data: Arc<dyn DataAdapter>,
projects: Arc<ProjectService>,
data_root: PathBuf,
}
pub const DEFAULT_HUB_SERVICE_INSTANCE_ID: &str = "hub-default";
pub const HUB_SERVICE_KIND: &str = "hub";
const HUB_SERVICE_SCOPE_OWNER: &str = "hub-service";
const HUB_SERVICE_SCOPE_PROJECT: &str = "hub-default";
const MAX_REMOTE_HUB_ARTIFACT_BYTES: u64 = 25 * 1024 * 1024;
const DEFAULT_PUBLISHER_MAX_PACKAGES: i64 = 20;
const DEFAULT_PUBLISHER_MAX_PACKAGE_BYTES: i64 = 10 * 1024 * 1024;
const DEFAULT_PUBLISHER_MAX_MEDIA_FILES: i64 = 8;
const DEFAULT_PUBLISHER_MAX_IMAGE_BYTES: i64 = 2 * 1024 * 1024;
const HUB_ASSET_KIND_PIPELINE_BUNDLE: &str = "pipeline_bundle";
const HUB_ASSET_KIND_TEMPLATE_BUNDLE: &str = "template_bundle";
const HUB_ASSET_KIND_FOLDER_BUNDLE: &str = "folder_bundle";
const HUB_ASSET_KIND_PROJECT_BUNDLE: &str = "project_bundle";
const HUB_ASSET_KIND_NODE_BUNDLE: &str = "node_bundle";
const HUB_ASSET_KINDS: &[&str] = &[
HUB_ASSET_KIND_PIPELINE_BUNDLE,
HUB_ASSET_KIND_TEMPLATE_BUNDLE,
HUB_ASSET_KIND_FOLDER_BUNDLE,
HUB_ASSET_KIND_PROJECT_BUNDLE,
HUB_ASSET_KIND_NODE_BUNDLE,
];
fn default_hub_base_url() -> String {
std::env::var("ZEBFLOW_HUB_DEFAULT_BASE_URL")
.ok()
.map(|v| v.trim().trim_end_matches('/').to_string())
.filter(|v| !v.is_empty())
.unwrap_or_else(|| "https://hub.zebflow.com/api".to_string())
}
fn preserve_or_replace_token(existing: Option<&str>, incoming: &str) -> String {
let trimmed = incoming.trim();
if trimmed.is_empty() {
existing.unwrap_or_default().to_string()
} else {
trimmed.to_string()
}
}
fn prune_empty_hub_dirs(start: Option<&Path>, stop: &Path) {
let Some(mut current) = start.map(Path::to_path_buf) else {
return;
};
while current.starts_with(stop) && current != stop {
match fs::remove_dir(¤t) {
Ok(()) => {}
Err(_) => break,
}
let Some(parent) = current.parent() else {
break;
};
current = parent.to_path_buf();
}
}
#[derive(Debug, Clone, Serialize)]
pub struct HubPublishSourceItem {
pub source_type: String,
pub source_ref: String,
pub name: String,
pub description: String,
pub path: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HubExportEntry {
pub rel_path: String,
pub kind: String,
pub size_bytes: usize,
pub reason: String,
#[serde(default)]
pub encoding: String,
#[serde(default)]
pub content: String,
}
#[derive(Debug, Clone, Serialize)]
pub struct HubExportPreview {
pub asset_kind: String,
pub source_type: String,
pub source_ref: String,
pub name: String,
pub description: String,
pub entries: Vec<HubExportEntry>,
pub warnings: Vec<String>,
pub total_files: usize,
pub total_bytes: usize,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct HubProjectBundlePublishOptions {
#[serde(default)]
pub include_sekejap_schema: bool,
#[serde(default)]
pub include_sqlite_schema: bool,
#[serde(default)]
pub include_libraries: Vec<String>,
#[serde(default)]
pub include_initial_data: bool,
#[serde(default)]
pub initial_data_paths: Vec<String>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
struct HubProjectInitialization {
#[serde(default)]
include_sekejap_schema: bool,
#[serde(default)]
include_sqlite_schema: bool,
#[serde(default)]
libraries: Vec<String>,
#[serde(default)]
initial_data: Vec<HubInitialDataStep>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct HubInitialDataStep {
pub engine: String,
pub path: String,
#[serde(default)]
pub statement_count: usize,
#[serde(default)]
pub size_bytes: usize,
}
#[derive(Debug, Clone, Serialize)]
pub struct HubInstallResult {
pub package_id: String,
pub version: String,
pub install_root: String,
pub files_written: usize,
pub pipelines_registered: Vec<String>,
}
#[derive(Debug, Clone, Serialize)]
pub struct HubInstallReview {
pub package_id: String,
pub version: String,
pub target_folder: String,
pub install_root: String,
pub asset_kind: String,
pub files_added: Vec<String>,
pub files_overwritten: Vec<String>,
pub pipelines_registered: Vec<String>,
pub nodes_used: Vec<String>,
pub credentials_required: Vec<String>,
pub external_urls: Vec<String>,
pub database_effects: Vec<String>,
pub filesystem_effects: Vec<String>,
pub public_endpoints: Vec<String>,
pub schedules: Vec<String>,
pub large_files: Vec<String>,
pub seed_data: Vec<String>,
pub project_initialization: serde_json::Value,
pub warnings: Vec<String>,
pub risk_level: String,
}
#[derive(Debug, Clone, Serialize)]
pub struct HubPublishReview {
pub package_id: String,
pub version: String,
pub asset_kind: String,
pub source_type: String,
pub source_ref: String,
pub title: String,
pub description: String,
pub visibility: String,
pub tags: Vec<String>,
pub total_files: usize,
pub total_bytes: usize,
pub files: Vec<HubExportEntry>,
pub media: Vec<HubPublishMediaReview>,
pub nodes_used: Vec<String>,
pub credentials_required: Vec<String>,
pub external_urls: Vec<String>,
pub database_effects: Vec<String>,
pub filesystem_effects: Vec<String>,
pub public_endpoints: Vec<String>,
pub schedules: Vec<String>,
pub large_files: Vec<String>,
pub seed_data: Vec<String>,
pub project_initialization: serde_json::Value,
pub warnings: Vec<String>,
pub risk_level: String,
}
#[derive(Debug, Clone, Serialize)]
pub struct HubPublishMediaReview {
pub name: String,
pub role: String,
pub content_type: String,
pub size_bytes: usize,
}
#[derive(Debug, Clone, Serialize)]
pub struct HubRemotePackRow {
pub repository_id: String,
pub repository_title: String,
pub package_id: String,
#[serde(skip_serializing_if = "String::is_empty")]
pub publisher_owner: String,
pub publisher_id: String,
pub publisher_display_name: String,
pub publisher_url: String,
#[serde(skip_serializing_if = "String::is_empty")]
pub publisher_email: String,
pub asset_kind: String,
pub title: String,
pub description: String,
pub summary: String,
pub image_url: String,
pub gallery: Value,
pub visibility: String,
pub tags: Vec<String>,
pub latest_version: String,
pub updated_at: i64,
pub source: String,
}
#[derive(Debug, Clone, Deserialize)]
struct RemoteHubListResponse {
items: Vec<RemoteHubAssetItem>,
}
#[derive(Debug, Clone, Deserialize)]
struct RemoteHubAssetItem {
package_id: String,
#[serde(default)]
publisher_owner: String,
#[serde(default)]
publisher_id: String,
#[serde(default)]
publisher_display_name: String,
#[serde(default)]
publisher_url: String,
#[serde(default)]
publisher_email: String,
asset_kind: String,
title: String,
description: String,
#[serde(default)]
summary: String,
#[serde(default)]
image_url: String,
#[serde(default)]
gallery: Value,
visibility: String,
#[serde(default)]
tags: Vec<String>,
#[serde(default)]
latest_version: String,
#[serde(default)]
updated_at: i64,
}
#[derive(Debug, Clone, Deserialize)]
struct RemoteHubArtifactResponse {
version: RemoteHubAssetVersion,
#[serde(default)]
artifact_sha256: String,
#[serde(default)]
artifact_size_bytes: u64,
artifact: Value,
}
#[derive(Debug, Clone, Deserialize)]
struct RemoteHubAssetVersion {
#[serde(default)]
artifact_sha256: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct HubArtifact {
schema: String,
asset_kind: String,
#[serde(default)]
source_type: String,
#[serde(default)]
source_owner: String,
#[serde(default)]
source_project: String,
#[serde(default)]
source_ref: String,
#[serde(default)]
publisher_id: String,
#[serde(default)]
publisher_display_name: String,
#[serde(default)]
publisher_url: String,
#[serde(default)]
publisher_email: String,
title: String,
description: String,
#[serde(default)]
summary: String,
#[serde(default)]
description_md: String,
#[serde(default)]
image_url: String,
#[serde(default)]
gallery: HubGallery,
#[serde(default)]
media: Vec<HubMediaFile>,
#[serde(default)]
active_pipelines: Vec<String>,
#[serde(default)]
project_initialization: HubProjectInitialization,
files: Vec<HubExportEntry>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HubMediaFile {
pub name: String,
pub role: String,
pub content_type: String,
pub size_bytes: usize,
pub sha256: String,
pub encoding: String,
pub content: String,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
struct HubGallery {
#[serde(default)]
cover: Option<HubGalleryImage>,
#[serde(default)]
items: Vec<HubGalleryItem>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct HubGalleryImage {
kind: String,
media_name: String,
#[serde(default)]
alt: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct HubGalleryItem {
kind: String,
#[serde(default)]
media_name: String,
#[serde(default)]
alt: String,
#[serde(default)]
url: String,
#[serde(default)]
title: String,
}
#[derive(Debug, Clone, serde::Serialize, Deserialize)]
pub struct RemoteHubPublishRequest {
pub package_id: String,
pub version: String,
#[serde(default)]
pub title: String,
#[serde(default)]
pub description: String,
#[serde(default)]
pub visibility: String,
#[serde(default)]
pub tags: Vec<String>,
pub source_owner: String,
pub source_project: String,
pub source_kind: String,
pub source_ref: String,
pub artifact: Value,
}
impl HubService {
pub fn new(
control_data: Arc<dyn DataAdapter>,
hub_data: Arc<dyn DataAdapter>,
projects: Arc<ProjectService>,
data_root: PathBuf,
) -> Self {
Self {
control_data,
hub_data,
projects,
data_root,
}
}
pub fn get_authority(
&self,
owner: &str,
project: &str,
) -> Result<Option<HubAuthority>, PlatformError> {
self.hub_data
.get_hub_authority(&slug_segment(owner), &slug_segment(project))
}
pub fn get_default_service_instance(
&self,
) -> Result<Option<PlatformServiceInstance>, PlatformError> {
self.control_data
.get_platform_service_instance(DEFAULT_HUB_SERVICE_INSTANCE_ID)
}
pub fn ensure_default_service_instance(
&self,
host_office_id: &str,
public_base_url: &str,
enabled: bool,
) -> Result<PlatformServiceInstance, PlatformError> {
let host_office_id = host_office_id.trim();
if host_office_id.is_empty() {
return Err(PlatformError::new(
"HUB_SERVICE_HOST_REQUIRED",
"hub service host office is required",
));
}
if self
.control_data
.get_platform_office(host_office_id)?
.is_none()
{
return Err(PlatformError::new(
"HUB_SERVICE_HOST_MISSING",
"hub service host office was not found",
));
}
let public_base_url = public_base_url.trim().trim_end_matches('/');
let public_base_url = if public_base_url.is_empty() {
default_hub_base_url()
} else {
public_base_url.to_string()
};
let existing = self.get_default_service_instance()?;
let placement_changed = existing.as_ref().is_some_and(|item| {
item.host_office_id != host_office_id
|| item.state_office_id != host_office_id
|| item.public_base_url != public_base_url
|| item.enabled != enabled
});
let now = now_ts();
let service = PlatformServiceInstance {
service_instance_id: DEFAULT_HUB_SERVICE_INSTANCE_ID.to_string(),
service_kind: HUB_SERVICE_KIND.to_string(),
display_label: "Hub".to_string(),
host_office_id: host_office_id.to_string(),
state_office_id: host_office_id.to_string(),
public_base_url,
enabled,
status: if enabled {
"enabled".to_string()
} else {
"disabled".to_string()
},
placement_generation: existing
.as_ref()
.map(|item| {
if placement_changed {
item.placement_generation.saturating_add(1)
} else {
item.placement_generation.max(1)
}
})
.unwrap_or(1),
created_at: existing.as_ref().map(|item| item.created_at).unwrap_or(now),
updated_at: now,
};
self.control_data.put_platform_service_instance(&service)?;
Ok(service)
}
pub fn set_authority_enabled(
&self,
owner: &str,
project: &str,
enabled: bool,
) -> Result<HubAuthority, PlatformError> {
let _ = (owner, project);
let authority = self.ensure_service_authority()?;
let now = now_ts();
let next = HubAuthority {
enabled,
updated_at: now,
..authority
};
self.hub_data.put_hub_authority(&next)?;
Ok(next)
}
pub fn list_asset_packages(&self) -> Result<Vec<HubAssetPackage>, PlatformError> {
self.require_enabled()?;
self.hub_data.list_hub_asset_packages()
}
pub fn list_asset_packages_by_owner(
&self,
owner: &str,
) -> Result<Vec<HubAssetPackage>, PlatformError> {
let owner = slug_segment(owner);
self.require_enabled()?;
let mut items = self.hub_data.list_hub_asset_packages()?;
items.retain(|item| item.publisher_owner == owner);
Ok(items)
}
pub fn list_asset_versions(
&self,
package_id: &str,
) -> Result<Vec<HubAssetVersion>, PlatformError> {
self.require_enabled()?;
self.hub_data.list_hub_asset_versions(package_id)
}
pub fn delete_asset_package(
&self,
token: &HubToken,
package_id: &str,
) -> Result<usize, PlatformError> {
self.require_enabled()?;
let package_id = canonical_hub_package_id(&token.publisher_id, package_id)?;
if package_id.is_empty() {
return Err(PlatformError::new(
"HUB_PACKAGE_INVALID",
"package id must not be empty",
));
}
let Some(package) = self.hub_data.get_hub_asset_package(&package_id)? else {
return Err(PlatformError::new(
"HUB_ASSET_MISSING",
"asset package not found",
));
};
let can_manage = token.scopes.iter().any(|scope| scope == "hub:manage");
if !can_manage
&& (package.publisher_owner != token.owner
|| package.publisher_id != token.publisher_id)
{
return Err(PlatformError::new(
"HUB_PACKAGE_FORBIDDEN",
"token cannot delete this package",
));
}
let versions = self.hub_data.list_hub_asset_versions(&package_id)?;
let artifact_paths = versions
.iter()
.filter_map(|version| {
self.hub_artifact_path_for_delete(&version.artifact_rel_path)
.ok()
})
.collect::<Vec<_>>();
self.hub_data.delete_hub_asset_package(&package_id)?;
for path in artifact_paths {
match fs::remove_file(&path) {
Ok(()) => {}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {}
Err(_) => {}
}
prune_empty_hub_dirs(path.parent(), &self.data_root);
}
Ok(versions.len())
}
pub fn list_publishers(
&self,
owner: &str,
project: &str,
) -> Result<Vec<HubPublisher>, PlatformError> {
let _ = (owner, project);
self.require_enabled()?;
self.hub_data
.list_hub_publishers(HUB_SERVICE_SCOPE_OWNER, HUB_SERVICE_SCOPE_PROJECT)
}
pub fn upsert_publisher(
&self,
owner: &str,
project: &str,
publisher_id: &str,
display_name: &str,
publisher_url: &str,
email: &str,
description: &str,
icon_url: &str,
website_url: &str,
enabled: bool,
can_read: bool,
can_publish: bool,
can_manage: bool,
max_packages: i64,
max_package_bytes: i64,
max_media_files: i64,
max_image_bytes: i64,
) -> Result<HubPublisher, PlatformError> {
let _ = (owner, project);
self.require_enabled()?;
let authority = self.ensure_service_authority()?;
let publisher_id = normalize_hub_id_segment(publisher_id, "publisher id")?;
if publisher_id.is_empty() {
return Err(PlatformError::new(
"HUB_PUBLISHER_INVALID",
"publisher id must not be empty",
));
}
let now = now_ts();
let existing = self.hub_data.get_hub_publisher(
HUB_SERVICE_SCOPE_OWNER,
HUB_SERVICE_SCOPE_PROJECT,
&publisher_id,
)?;
let row = HubPublisher {
authority_id: authority.authority_id,
publisher_pk: existing
.as_ref()
.map(|v| v.publisher_pk.clone())
.filter(|v| !v.is_empty())
.unwrap_or_else(|| format!("mpub_{}", random_hex(8))),
owner: HUB_SERVICE_SCOPE_OWNER.to_string(),
project: HUB_SERVICE_SCOPE_PROJECT.to_string(),
publisher_id: publisher_id.clone(),
display_name: if display_name.trim().is_empty() {
publisher_id.replace('-', " ")
} else {
display_name.trim().to_string()
},
publisher_url: normalize_publisher_url(&publisher_id, publisher_url),
email: email.trim().to_string(),
description: description.trim().to_string(),
icon_url: icon_url.trim().to_string(),
website_url: website_url.trim().to_string(),
enabled,
can_read,
can_publish,
can_manage,
max_packages: normalize_limit(max_packages, DEFAULT_PUBLISHER_MAX_PACKAGES),
max_package_bytes: normalize_limit(
max_package_bytes,
DEFAULT_PUBLISHER_MAX_PACKAGE_BYTES,
),
max_media_files: normalize_limit(max_media_files, DEFAULT_PUBLISHER_MAX_MEDIA_FILES),
max_image_bytes: normalize_limit(max_image_bytes, DEFAULT_PUBLISHER_MAX_IMAGE_BYTES),
created_at: existing.as_ref().map(|v| v.created_at).unwrap_or(now),
updated_at: now,
};
self.hub_data.put_hub_publisher(&row)?;
Ok(row)
}
pub fn delete_publisher(
&self,
owner: &str,
project: &str,
publisher_id: &str,
) -> Result<(), PlatformError> {
let _ = (owner, project);
self.hub_data.delete_hub_publisher(
HUB_SERVICE_SCOPE_OWNER,
HUB_SERVICE_SCOPE_PROJECT,
&slug_segment(publisher_id),
)
}
pub fn create_token(
&self,
owner: &str,
project: &str,
req: &CreateHubTokenRequest,
) -> Result<(HubToken, String), PlatformError> {
let owner = slug_segment(owner);
let project = slug_segment(project);
self.require_enabled()?;
let authority = self.ensure_service_authority()?;
if owner.is_empty() || project.is_empty() {
return Err(PlatformError::new(
"HUB_TOKEN_INVALID",
"owner/project must not be empty",
));
}
let publisher_id = normalize_hub_id_segment(&req.publisher_id, "publisher id")?;
let title = req.title.trim();
if title.is_empty() || publisher_id.is_empty() {
return Err(PlatformError::new(
"HUB_TOKEN_INVALID",
"token title and publisher id must not be empty",
));
}
let Some(publisher) = self.hub_data.get_hub_publisher(
HUB_SERVICE_SCOPE_OWNER,
HUB_SERVICE_SCOPE_PROJECT,
&publisher_id,
)?
else {
return Err(PlatformError::new(
"HUB_PUBLISHER_MISSING",
"publisher not found",
));
};
if !publisher.enabled {
return Err(PlatformError::new(
"HUB_PUBLISHER_DISABLED",
"publisher is disabled",
));
}
let scopes = normalize_scopes(&req.scopes);
validate_publisher_scopes(&publisher, &scopes)?;
let now = now_ts();
let token_id = format!("mkt_{}", random_hex(8));
let secret = random_hex(24);
let plain = format!("zfmt_{token_id}_{secret}");
let token = HubToken {
token_id: token_id.clone(),
authority_id: authority.authority_id,
publisher_pk: publisher.publisher_pk.clone(),
owner,
project,
publisher_id: publisher_id.clone(),
publisher_display_name: publisher.display_name,
publisher_url: publisher.publisher_url,
publisher_email: publisher.email,
title: title.to_string(),
secret_hash: sha256_hex(plain.as_bytes()),
scopes,
scope_read: false,
scope_publish: false,
scope_manage: false,
expires_at: req.expires_at,
last_used_at: None,
revoked_at: None,
created_at: now,
updated_at: now,
};
let token = apply_scope_flags(token);
self.hub_data.put_hub_token(&token)?;
Ok((token, plain))
}
pub fn list_tokens(&self, owner: &str, project: &str) -> Result<Vec<HubToken>, PlatformError> {
let owner = slug_segment(owner);
let project = slug_segment(project);
self.require_enabled()?;
self.hub_data.list_hub_tokens(&owner, &project)
}
pub fn list_all_tokens(&self) -> Result<Vec<HubToken>, PlatformError> {
self.require_enabled()?;
self.hub_data.list_all_hub_tokens()
}
pub fn revoke_token(
&self,
owner: &str,
project: &str,
token_id: &str,
) -> Result<(), PlatformError> {
let owner = slug_segment(owner);
let project = slug_segment(project);
let Some(mut token) = self.hub_data.get_hub_token(token_id)? else {
return Ok(());
};
if token.owner != owner || token.project != project {
return Err(PlatformError::new(
"HUB_TOKEN_FORBIDDEN",
"token does not belong to this hub",
));
}
token.revoked_at = Some(now_ts());
token.updated_at = now_ts();
self.hub_data.put_hub_token(&token)?;
Ok(())
}
pub fn revoke_token_any(&self, token_id: &str) -> Result<(), PlatformError> {
self.require_enabled()?;
let Some(mut token) = self.hub_data.get_hub_token(token_id)? else {
return Ok(());
};
token.revoked_at = Some(now_ts());
token.updated_at = now_ts();
self.hub_data.put_hub_token(&token)?;
Ok(())
}
pub fn list_repositories(
&self,
owner: &str,
project: &str,
) -> Result<Vec<ProjectHubRepository>, PlatformError> {
self.control_data
.list_project_hub_repositories(&slug_segment(owner), &slug_segment(project))
}
pub fn list_access_grants(
&self,
source_owner: &str,
) -> Result<Vec<HubAccessGrant>, PlatformError> {
self.control_data
.list_hub_access_grants(&slug_segment(source_owner))
}
pub fn upsert_access_grant(
&self,
source_owner: &str,
repository_id: &str,
grant_scope: &str,
target_owner: &str,
target_project: &str,
can_read: bool,
can_publish: bool,
can_manage: bool,
enabled: bool,
) -> Result<HubAccessGrant, PlatformError> {
let source_owner = slug_segment(source_owner);
let repository_id = slug_segment(repository_id);
let grant_scope = normalize_hub_grant_scope(grant_scope)?;
let target_owner = if grant_scope == "all_projects" {
String::new()
} else {
slug_segment(target_owner)
};
let target_project = if grant_scope == "all_projects" {
String::new()
} else {
slug_segment(target_project)
};
if source_owner.is_empty()
|| repository_id.is_empty()
|| (grant_scope == "selected_project"
&& (target_owner.is_empty() || target_project.is_empty()))
{
return Err(PlatformError::new(
"HUB_ACCESS_GRANT_INVALID",
"source owner, repository id, and selected project target are required",
));
}
let repos = self
.control_data
.list_platform_hub_repositories(&source_owner)?;
let Some(repo) = repos
.into_iter()
.find(|item| item.repository_id == repository_id)
else {
return Err(PlatformError::new(
"HUB_REPOSITORY_MISSING",
"platform Hub source not found",
));
};
let now = now_ts();
let existing = self
.control_data
.list_hub_access_grants(&source_owner)?
.into_iter()
.find(|item| {
item.source_id == repo.source_id
&& item.grant_scope == grant_scope
&& item.target_owner == target_owner
&& item.target_project == target_project
});
let grant = HubAccessGrant {
grant_id: existing
.as_ref()
.map(|item| item.grant_id.clone())
.filter(|value| !value.is_empty())
.unwrap_or_else(|| format!("hgrant_{}", random_hex(8))),
source_owner,
source_id: repo.source_id,
repository_id,
grant_scope,
target_owner,
target_project,
can_read,
can_publish,
can_manage,
enabled,
created_at: existing.as_ref().map(|item| item.created_at).unwrap_or(now),
updated_at: now,
};
self.control_data.put_hub_access_grant(&grant)?;
Ok(grant)
}
pub fn delete_access_grant(&self, grant_id: &str) -> Result<(), PlatformError> {
self.control_data
.delete_hub_access_grant(&slug_segment(grant_id))
}
pub fn list_effective_repositories(
&self,
owner: &str,
project: &str,
) -> Result<Vec<ProjectHubRepository>, PlatformError> {
let owner = slug_segment(owner);
let project = slug_segment(project);
let mut out = self
.control_data
.list_project_hub_repositories(&owner, &project)?;
let mut seen = out
.iter()
.map(|item| item.repository_id.clone())
.collect::<BTreeSet<_>>();
let grants = self
.control_data
.list_effective_hub_access_grants(&owner, &project)?;
for grant in grants
.into_iter()
.filter(|item| item.enabled && item.can_read)
{
if seen.contains(&grant.repository_id) {
continue;
}
let source = self
.control_data
.list_platform_hub_repositories(&grant.source_owner)?
.into_iter()
.find(|item| {
item.source_id == grant.source_id || item.repository_id == grant.repository_id
});
let Some(source) = source else {
continue;
};
if !source.enabled {
continue;
}
seen.insert(source.repository_id.clone());
out.push(ProjectHubRepository {
owner: owner.clone(),
project: project.clone(),
repository_id: source.repository_id,
title: source.title,
base_url: source.base_url,
remote_owner: source.remote_owner,
remote_project: source.remote_project,
read_token: source.read_token,
enabled: source.enabled,
created_at: grant.created_at,
updated_at: grant.updated_at,
});
}
Ok(out)
}
pub fn ensure_default_project_repository(
&self,
owner: &str,
project: &str,
) -> Result<(), PlatformError> {
let owner = slug_segment(owner);
let project = slug_segment(project);
if owner.is_empty() || project.is_empty() {
return Ok(());
}
self.ensure_default_platform_repository(&owner)?;
let has_default_grant = self
.control_data
.list_hub_access_grants(&owner)?
.into_iter()
.any(|grant| {
grant.repository_id == "zebflow-com"
&& grant.grant_scope == "all_projects"
&& grant.enabled
&& grant.can_read
});
if !has_default_grant {
self.upsert_access_grant(
&owner,
"zebflow-com",
"all_projects",
"",
"",
true,
false,
false,
true,
)?;
}
Ok(())
}
pub fn list_platform_repositories(
&self,
owner: &str,
) -> Result<Vec<PlatformHubRepository>, PlatformError> {
self.control_data
.list_platform_hub_repositories(&slug_segment(owner))
}
pub fn upsert_platform_repository(
&self,
owner: &str,
repository_id: &str,
title: &str,
base_url: &str,
remote_owner: &str,
remote_project: &str,
read_token: &str,
visibility: &str,
enabled: bool,
) -> Result<PlatformHubRepository, PlatformError> {
let owner = slug_segment(owner);
let repository_id = slug_segment(repository_id);
if owner.is_empty() || repository_id.is_empty() || base_url.trim().is_empty() {
return Err(PlatformError::new(
"HUB_REPOSITORY_INVALID",
"repository fields must not be empty",
));
}
validate_remote_hub_url(base_url.trim().trim_end_matches('/'))?;
let uses_direct_base = is_direct_hub_base(base_url);
if !uses_direct_base && (remote_owner.trim().is_empty() || remote_project.trim().is_empty())
{
return Err(PlatformError::new(
"HUB_REPOSITORY_INVALID",
"remote owner and project are required unless base_url already points to /api/projects/{owner}/{project}/hub",
));
}
let now = now_ts();
let owner_user_id = self
.control_data
.get_user_auth(&owner)?
.map(|user| user.profile.user_id)
.ok_or_else(|| PlatformError::new("PLATFORM_USER_NOT_FOUND", "owner user not found"))?;
let existing = self
.control_data
.list_platform_hub_repositories(&owner)?
.into_iter()
.find(|item| item.repository_id == repository_id);
let row = PlatformHubRepository {
source_id: existing
.as_ref()
.map(|item| item.source_id.clone())
.filter(|value| !value.is_empty())
.unwrap_or_else(|| format!("pmr_{}", random_hex(8))),
owner_user_id: existing
.as_ref()
.map(|item| item.owner_user_id.clone())
.filter(|value| !value.is_empty())
.unwrap_or(owner_user_id),
owner,
repository_id,
title: if title.trim().is_empty() {
"Remote Hub".to_string()
} else {
title.trim().to_string()
},
base_url: base_url.trim().trim_end_matches('/').to_string(),
remote_owner: if uses_direct_base {
String::new()
} else {
slug_segment(remote_owner)
},
remote_project: if uses_direct_base {
String::new()
} else {
slug_segment(remote_project)
},
read_token: preserve_or_replace_token(
existing.as_ref().map(|item| item.read_token.as_str()),
read_token,
),
visibility: normalize_platform_source_visibility(visibility),
enabled,
created_at: existing.as_ref().map(|item| item.created_at).unwrap_or(now),
updated_at: now,
};
self.control_data.put_platform_hub_repository(&row)?;
Ok(row)
}
pub fn delete_platform_repository(
&self,
owner: &str,
repository_id: &str,
) -> Result<(), PlatformError> {
self.control_data
.delete_platform_hub_repository(&slug_segment(owner), &slug_segment(repository_id))
}
pub fn ensure_default_platform_repository(&self, owner: &str) -> Result<(), PlatformError> {
let owner = slug_segment(owner);
if owner.is_empty() {
return Ok(());
}
let existing = self.control_data.list_platform_hub_repositories(&owner)?;
if existing
.iter()
.any(|item| item.repository_id == "zebflow-com")
{
return Ok(());
}
let now = now_ts();
let owner_user_id = self
.control_data
.get_user_auth(&owner)?
.map(|user| user.profile.user_id)
.ok_or_else(|| PlatformError::new("PLATFORM_USER_NOT_FOUND", "owner user not found"))?;
self.control_data
.put_platform_hub_repository(&PlatformHubRepository {
source_id: format!("pmr_{}", random_hex(8)),
owner_user_id,
owner,
repository_id: "zebflow-com".to_string(),
title: "Zebflow Hub".to_string(),
base_url: default_hub_base_url(),
remote_owner: String::new(),
remote_project: String::new(),
read_token: String::new(),
visibility: "public".to_string(),
enabled: true,
created_at: now,
updated_at: now,
})?;
Ok(())
}
pub fn upsert_repository(
&self,
owner: &str,
project: &str,
repository_id: &str,
title: &str,
base_url: &str,
remote_owner: &str,
remote_project: &str,
read_token: &str,
enabled: bool,
) -> Result<ProjectHubRepository, PlatformError> {
let owner = slug_segment(owner);
let project = slug_segment(project);
let repository_id = slug_segment(repository_id);
if owner.is_empty()
|| project.is_empty()
|| repository_id.is_empty()
|| base_url.trim().is_empty()
{
return Err(PlatformError::new(
"HUB_REPOSITORY_INVALID",
"repository fields must not be empty",
));
}
validate_remote_hub_url(base_url.trim().trim_end_matches('/'))?;
let uses_direct_base = is_direct_hub_base(base_url);
if !uses_direct_base && (remote_owner.trim().is_empty() || remote_project.trim().is_empty())
{
return Err(PlatformError::new(
"HUB_REPOSITORY_INVALID",
"remote owner and project are required unless base_url already points to /api/projects/{owner}/{project}/hub",
));
}
let now = now_ts();
let existing = self
.control_data
.list_project_hub_repositories(&owner, &project)?
.into_iter()
.find(|item| item.repository_id == repository_id);
let row = ProjectHubRepository {
owner,
project,
repository_id,
title: if title.trim().is_empty() {
"Remote Hub".to_string()
} else {
title.trim().to_string()
},
base_url: base_url.trim().trim_end_matches('/').to_string(),
remote_owner: if uses_direct_base {
String::new()
} else {
slug_segment(remote_owner)
},
remote_project: if uses_direct_base {
String::new()
} else {
slug_segment(remote_project)
},
read_token: preserve_or_replace_token(
existing.as_ref().map(|item| item.read_token.as_str()),
read_token,
),
enabled,
created_at: existing.as_ref().map(|item| item.created_at).unwrap_or(now),
updated_at: now,
};
self.control_data.put_project_hub_repository(&row)?;
Ok(row)
}
pub fn delete_repository(
&self,
owner: &str,
project: &str,
repository_id: &str,
) -> Result<(), PlatformError> {
self.control_data.delete_project_hub_repository(
&slug_segment(owner),
&slug_segment(project),
&slug_segment(repository_id),
)
}
pub fn authenticate_token(
&self,
bearer_token: &str,
required_scope: &str,
) -> Result<HubToken, PlatformError> {
self.require_enabled()?;
let token_value = bearer_token.trim();
if token_value.is_empty() {
return Err(PlatformError::new("HUB_TOKEN_INVALID", "token missing"));
}
let token_hash = sha256_hex(token_value.as_bytes());
let matched = self
.hub_data
.list_all_hub_tokens()?
.into_iter()
.find(|token| token.secret_hash == token_hash);
let Some(mut token) = matched else {
return Err(PlatformError::new("HUB_TOKEN_INVALID", "token not found"));
};
if token.revoked_at.is_some() {
return Err(PlatformError::new("HUB_TOKEN_REVOKED", "token revoked"));
}
if let Some(expires_at) = token.expires_at
&& expires_at > 0
&& expires_at <= now_ts()
{
return Err(PlatformError::new("HUB_TOKEN_EXPIRED", "token expired"));
}
if !token.grants_scope(required_scope) {
return Err(PlatformError::new("HUB_TOKEN_FORBIDDEN", "scope missing"));
}
let Some(publisher) = self.hub_data.get_hub_publisher(
HUB_SERVICE_SCOPE_OWNER,
HUB_SERVICE_SCOPE_PROJECT,
&token.publisher_id,
)?
else {
return Err(PlatformError::new(
"HUB_PUBLISHER_MISSING",
"publisher not found",
));
};
if !publisher.enabled {
return Err(PlatformError::new(
"HUB_PUBLISHER_DISABLED",
"publisher is disabled",
));
}
validate_publisher_scopes(&publisher, &[required_scope.to_string()])?;
token.last_used_at = Some(now_ts());
token.updated_at = now_ts();
self.hub_data.put_hub_token(&token)?;
Ok(token)
}
pub fn list_publish_sources(
&self,
owner: &str,
project: &str,
source_type: &str,
) -> Result<Vec<HubPublishSourceItem>, PlatformError> {
let owner = slug_segment(owner);
let project = slug_segment(project);
let source_type = normalize_source_type(source_type);
match source_type.as_str() {
"pipeline_with_dependencies" => {
let rows = self.projects.list_pipeline_meta_rows(&owner, &project)?;
Ok(rows
.into_iter()
.map(|item| HubPublishSourceItem {
source_type: source_type.clone(),
source_ref: item.file_rel_path.clone(),
name: item.title.clone(),
description: if item.description.trim().is_empty() {
format!("Pipeline · {}", item.trigger_kind)
} else {
item.description.clone()
},
path: item.file_rel_path,
})
.collect())
}
"template_with_dependencies" => {
let listing = self.projects.list_template_workspace(&owner, &project)?;
Ok(listing
.items
.into_iter()
.filter(|item| item.kind != "folder")
.map(|item| HubPublishSourceItem {
source_type: source_type.clone(),
source_ref: item.rel_path.clone(),
name: item.name,
description: format!("Template · {}", item.file_kind),
path: item.rel_path,
})
.collect())
}
"folder_files" => {
let layout = self.projects.project_layout(&owner, &project)?;
let folders = list_repo_folders(&layout)?;
Ok(folders
.into_iter()
.map(|path| HubPublishSourceItem {
source_type: source_type.clone(),
source_ref: path.clone(),
name: Path::new(&path)
.file_name()
.and_then(|v| v.to_str())
.unwrap_or(&path)
.to_string(),
description: "Folder files export".to_string(),
path,
})
.collect())
}
"project_files" => Ok(vec![HubPublishSourceItem {
source_type,
source_ref: ".".to_string(),
name: format!("{project} project files"),
description: "Entire repo workspace export".to_string(),
path: "/".to_string(),
}]),
_ => Err(PlatformError::new(
"HUB_SOURCE_INVALID",
"unsupported source type",
)),
}
}
pub fn preview_publish_source(
&self,
owner: &str,
project: &str,
source_type: &str,
source_ref: &str,
) -> Result<HubExportPreview, PlatformError> {
let owner = slug_segment(owner);
let project = slug_segment(project);
let source_type = normalize_source_type(source_type);
let layout = self.projects.project_layout(&owner, &project)?;
match source_type.as_str() {
"pipeline_with_dependencies" => {
self.preview_pipeline(&owner, &project, &layout, source_ref)
}
"template_with_dependencies" => {
self.preview_template(&owner, &project, &layout, source_ref)
}
"folder_files" => self.preview_folder(&layout, source_ref),
"project_files" => self.preview_project(&layout),
_ => Err(PlatformError::new(
"HUB_SOURCE_INVALID",
"unsupported source type",
)),
}
}
pub fn publish_asset(
&self,
authority_owner: &str,
authority_project: &str,
publisher_owner: &str,
publisher_id: &str,
publisher_display_name: &str,
publisher_url: &str,
publisher_email: &str,
source_owner: &str,
source_project: &str,
source_type: &str,
source_ref: &str,
package_id: &str,
version: &str,
title: &str,
description: &str,
image_file_path: &str,
visibility: &str,
project_options: HubProjectBundlePublishOptions,
tags: Vec<String>,
) -> Result<(HubAssetPackage, HubAssetVersion), PlatformError> {
let _ = (authority_owner, authority_project);
self.require_enabled()?;
let authority_owner = HUB_SERVICE_SCOPE_OWNER.to_string();
let authority_project = HUB_SERVICE_SCOPE_PROJECT.to_string();
let publisher_owner = slug_segment(publisher_owner);
let publisher_id = normalize_hub_id_segment(publisher_id, "publisher id")?;
let source_owner = slug_segment(source_owner);
let source_project = slug_segment(source_project);
let source_type = normalize_source_type(source_type);
let package_id = canonical_hub_package_id(&publisher_id, package_id)?;
let version = version.trim();
if authority_owner.is_empty()
|| authority_project.is_empty()
|| publisher_owner.is_empty()
|| publisher_id.is_empty()
|| source_owner.is_empty()
|| source_project.is_empty()
{
return Err(PlatformError::new(
"HUB_PUBLISH_INVALID",
"authority, publisher, and source must not be empty",
));
}
let Some(publisher) = self.hub_data.get_hub_publisher(
HUB_SERVICE_SCOPE_OWNER,
HUB_SERVICE_SCOPE_PROJECT,
&publisher_id,
)?
else {
return Err(PlatformError::new(
"HUB_PUBLISHER_MISSING",
"publisher not found",
));
};
if !publisher.enabled {
return Err(PlatformError::new(
"HUB_PUBLISHER_DISABLED",
"publisher is disabled",
));
}
if !publisher.can_publish {
return Err(PlatformError::new(
"HUB_PUBLISHER_FORBIDDEN",
"publisher does not have publish permission",
));
}
let authority = self.ensure_service_authority()?;
if package_id.is_empty() || version.is_empty() {
return Err(PlatformError::new(
"HUB_PUBLISH_INVALID",
"package id and version must not be empty",
));
}
validate_hub_version(version)?;
let mut preview =
self.preview_publish_source(&source_owner, &source_project, &source_type, source_ref)?;
if preview.entries.is_empty() {
return Err(PlatformError::new(
"HUB_PUBLISH_EMPTY",
"nothing to publish for this source",
));
}
let project_initialization = self.apply_project_bundle_publish_options(
&source_owner,
&source_project,
&source_type,
&mut preview,
project_options,
)?;
sanitize_hub_export_entries(&mut preview.entries)?;
let media = collect_publish_media_from_files(
&self
.projects
.project_layout(&source_owner, &source_project)?,
image_file_path,
&publisher,
)?;
let active_pipelines = if preview.asset_kind == HUB_ASSET_KIND_PROJECT_BUNDLE {
self.projects
.list_active_pipeline_meta(&source_owner, &source_project)?
.into_iter()
.map(|meta| meta.file_rel_path)
.collect()
} else {
Vec::new()
};
let image_url = media
.iter()
.find(|item| item.role == "cover")
.map(|item| format!("/api/hub/remote/assets/{package_id}/media/{}", item.name))
.unwrap_or_default();
let artifact_rel = format!(
"services/{}/packages/{}/versions/{}/artifact.json",
DEFAULT_HUB_SERVICE_INSTANCE_ID, package_id, version
);
let artifact_abs = self.data_root.join(&artifact_rel);
if let Some(parent) = artifact_abs.parent() {
fs::create_dir_all(parent)?;
}
let now = now_ts();
let manifest = HubArtifact {
schema: "zebflow.asset-pack.v1".to_string(),
asset_kind: preview.asset_kind.clone(),
source_type: source_type.clone(),
source_owner: source_owner.clone(),
source_project: source_project.clone(),
source_ref: source_ref.to_string(),
publisher_id: publisher_id.clone(),
publisher_display_name: if publisher_display_name.trim().is_empty() {
publisher.display_name.clone()
} else {
publisher_display_name.trim().to_string()
},
publisher_url: if publisher_url.trim().is_empty() {
publisher.publisher_url.clone()
} else {
normalize_publisher_url(&publisher_id, publisher_url)
},
publisher_email: if publisher_email.trim().is_empty() {
publisher.email.clone()
} else {
publisher_email.trim().to_string()
},
title: if title.trim().is_empty() {
preview.name.clone()
} else {
title.trim().to_string()
},
description: if description.trim().is_empty() {
preview.description.clone()
} else {
description.trim().to_string()
},
summary: String::new(),
description_md: String::new(),
image_url: image_url.clone(),
gallery: HubGallery::default(),
media,
active_pipelines,
project_initialization,
files: preview.entries.clone(),
};
let artifact_bytes = serde_json::to_vec_pretty(&manifest)
.map_err(|err| PlatformError::new("HUB_PUBLISH", err.to_string()))?;
let artifact_sha256 = sha256_hex(&artifact_bytes);
let existing_package = self.hub_data.get_hub_asset_package(&package_id)?;
self.enforce_publisher_package_quota(
&publisher,
&package_id,
existing_package.as_ref(),
artifact_bytes.len(),
)?;
fs::write(&artifact_abs, &artifact_bytes)?;
let package = HubAssetPackage {
package_pk: existing_package
.as_ref()
.map(|item| item.package_pk.clone())
.filter(|value| !value.is_empty())
.unwrap_or_else(|| format!("mpkg_{}", random_hex(8))),
authority_id: authority.authority_id.clone(),
publisher_pk: publisher.publisher_pk.clone(),
package_id: package_id.clone(),
authority_owner: authority_owner.clone(),
authority_project: authority_project.clone(),
publisher_owner: publisher_owner.clone(),
publisher_id: publisher_id.clone(),
publisher_display_name: manifest.publisher_display_name.clone(),
publisher_url: manifest.publisher_url.clone(),
publisher_email: manifest.publisher_email.clone(),
asset_kind: preview.asset_kind.clone(),
title: manifest.title.clone(),
description: manifest.description.clone(),
image_url,
visibility: normalize_visibility(visibility),
tags,
created_at: existing_package.map(|item| item.created_at).unwrap_or(now),
updated_at: now,
};
let version_row = HubAssetVersion {
package_pk: package.package_pk.clone(),
package_id: package_id.clone(),
version: version.to_string(),
authority_owner,
authority_project,
publisher_owner,
publisher_id,
source_owner,
source_project,
source_kind: source_type,
source_ref: source_ref.to_string(),
artifact_rel_path: artifact_rel,
artifact_sha256,
manifest: serde_json::to_value(&manifest)
.map_err(|err| PlatformError::new("HUB_PUBLISH", err.to_string()))?,
created_at: now,
};
self.hub_data.put_hub_asset_package(&package)?;
self.hub_data.put_hub_asset_version(&version_row)?;
Ok((package, version_row))
}
pub fn review_publish_asset(
&self,
source_owner: &str,
source_project: &str,
publisher_id: &str,
source_type: &str,
source_ref: &str,
package_id: &str,
version: &str,
title: &str,
description: &str,
image_file_path: &str,
visibility: &str,
project_options: HubProjectBundlePublishOptions,
tags: Vec<String>,
) -> Result<HubPublishReview, PlatformError> {
self.require_enabled()?;
let source_owner = slug_segment(source_owner);
let source_project = slug_segment(source_project);
let publisher_id = normalize_hub_id_segment(publisher_id, "publisher id")?;
let source_type = normalize_source_type(source_type);
let package_id = canonical_hub_package_id(&publisher_id, package_id)?;
let version = version.trim().to_string();
validate_hub_version(&version)?;
let Some(publisher) = self.hub_data.get_hub_publisher(
HUB_SERVICE_SCOPE_OWNER,
HUB_SERVICE_SCOPE_PROJECT,
&publisher_id,
)?
else {
return Err(PlatformError::new(
"HUB_PUBLISHER_MISSING",
"publisher not found",
));
};
let mut preview =
self.preview_publish_source(&source_owner, &source_project, &source_type, source_ref)?;
let project_initialization = self.apply_project_bundle_publish_options(
&source_owner,
&source_project,
&source_type,
&mut preview,
project_options,
)?;
sanitize_hub_export_entries(&mut preview.entries)?;
let media = collect_publish_media_from_files(
&self
.projects
.project_layout(&source_owner, &source_project)?,
image_file_path,
&publisher,
)?;
let title = if title.trim().is_empty() {
preview.name.clone()
} else {
title.trim().to_string()
};
let description = if description.trim().is_empty() {
preview.description.clone()
} else {
description.trim().to_string()
};
review_publish_artifact(
package_id,
version,
preview,
title,
description,
normalize_visibility(visibility),
tags,
media,
project_initialization,
)
}
fn apply_project_bundle_publish_options(
&self,
source_owner: &str,
source_project: &str,
source_type: &str,
preview: &mut HubExportPreview,
mut options: HubProjectBundlePublishOptions,
) -> Result<HubProjectInitialization, PlatformError> {
if source_type != "project_files" || preview.asset_kind != HUB_ASSET_KIND_PROJECT_BUNDLE {
return Ok(HubProjectInitialization::default());
}
options.include_libraries.sort();
options.include_libraries.dedup();
options.initial_data_paths = options
.initial_data_paths
.into_iter()
.map(|path| normalize_repo_rel(&path))
.filter(|path| initial_data_engine_for_path(path).is_some())
.collect();
options.initial_data_paths.sort();
options.initial_data_paths.dedup();
if !options.include_sekejap_schema {
preview.entries.retain(|entry| {
!normalize_repo_rel(&entry.rel_path).starts_with("schemas/sekejap/")
});
} else if !preview
.entries
.iter()
.any(|entry| normalize_repo_rel(&entry.rel_path) == "schemas/sekejap/schema.json")
{
let export = sekejap::export_schema(&self.data_root, source_owner, source_project)?;
if !export.tables.is_empty() {
let content = serde_json::to_string_pretty(&export)
.map_err(|err| PlatformError::new("HUB_PUBLISH", err.to_string()))?
+ "\n";
preview.entries.push(text_export_entry(
"schemas/sekejap/schema.json",
"sekejap schema",
"Portable Sekejap schema",
content,
));
}
}
preview
.entries
.retain(|entry| normalize_repo_rel(&entry.rel_path) != sqlite_schema::REPO_SCHEMA_PATH);
if options.include_sqlite_schema {
if let Some(sql) =
sqlite_schema::export_schema_sql(&self.data_root, source_owner, source_project)?
{
preview.entries.push(text_export_entry(
sqlite_schema::REPO_SCHEMA_PATH,
"sqlite schema",
"Portable SQLite schema",
sql,
));
}
}
let initial_data = if options.include_initial_data {
preview
.entries
.iter()
.filter_map(initial_data_step_from_entry)
.filter(|step| {
options
.initial_data_paths
.iter()
.any(|path| path == &step.path)
})
.collect::<Vec<_>>()
} else {
Vec::new()
};
rewrite_project_libraries(preview, &options.include_libraries)?;
let init = HubProjectInitialization {
include_sekejap_schema: options.include_sekejap_schema,
include_sqlite_schema: options.include_sqlite_schema,
libraries: options.include_libraries,
initial_data,
};
let init_content = serde_json::to_string_pretty(&init)
.map_err(|err| PlatformError::new("HUB_PUBLISH", err.to_string()))?
+ "\n";
preview
.entries
.retain(|entry| normalize_repo_rel(&entry.rel_path) != "zebflow.init.json");
preview.entries.push(text_export_entry(
"zebflow.init.json",
"project initialization",
"Project bundle initialization plan",
init_content,
));
Ok(init)
}
pub fn list_project_initial_data(
&self,
owner: &str,
project: &str,
) -> Result<Vec<HubInitialDataStep>, PlatformError> {
let layout = self.projects.project_layout(owner, project)?;
let mut steps = Vec::new();
for (prefix, engine) in INITIAL_DATA_DIRS {
let root = layout.repo_dir.join(prefix);
if !root.is_dir() {
continue;
}
collect_initial_data_steps(&layout.repo_dir, &root, engine, &mut steps)?;
}
steps.sort_by(|a, b| a.path.cmp(&b.path));
steps.dedup_by(|a, b| a.path == b.path);
Ok(steps)
}
pub fn install_asset(
&self,
target_owner: &str,
target_project: &str,
package_id: &str,
version: &str,
target_folder: &str,
) -> Result<HubInstallResult, PlatformError> {
let target_owner = slug_segment(target_owner);
let target_project = slug_segment(target_project);
self.require_enabled()?;
let Some(version_row) = self.hub_data.get_hub_asset_version(package_id, version)? else {
return Err(PlatformError::new(
"HUB_ASSET_MISSING",
"asset version not found",
));
};
let artifact_abs = self.data_root.join(&version_row.artifact_rel_path);
let raw = fs::read_to_string(&artifact_abs)?;
let payload: HubArtifact = serde_json::from_str(&raw)
.map_err(|err| PlatformError::new("HUB_INSTALL", err.to_string()))?;
self.install_artifact_payload(
target_owner,
target_project,
package_id,
version,
target_folder,
payload,
)
}
pub fn review_asset_install(
&self,
target_owner: &str,
target_project: &str,
package_id: &str,
version: &str,
target_folder: &str,
) -> Result<HubInstallReview, PlatformError> {
let target_owner = slug_segment(target_owner);
let target_project = slug_segment(target_project);
self.require_enabled()?;
let Some(version_row) = self.hub_data.get_hub_asset_version(package_id, version)? else {
return Err(PlatformError::new(
"HUB_ASSET_MISSING",
"asset version not found",
));
};
let artifact_abs = self.data_root.join(&version_row.artifact_rel_path);
let raw = fs::read_to_string(&artifact_abs)?;
let payload: HubArtifact = serde_json::from_str(&raw)
.map_err(|err| PlatformError::new("HUB_INSTALL_REVIEW", err.to_string()))?;
self.review_artifact_payload(
&target_owner,
&target_project,
package_id,
version,
target_folder,
&payload,
)
}
fn install_artifact_payload(
&self,
target_owner: String,
target_project: String,
package_id: &str,
version: &str,
target_folder: &str,
payload: HubArtifact,
) -> Result<HubInstallResult, PlatformError> {
let layout = self
.projects
.project_layout(&target_owner, &target_project)?;
let install_root =
install_root_for_target_folder(package_id, &payload.asset_kind, target_folder);
let mut pipelines_registered = Vec::new();
for entry in &payload.files {
let install_rel = install_rel_path_under_folder(&install_root, &entry.rel_path);
let dest_abs = layout.repo_dir.join(&install_rel);
if let Some(parent) = dest_abs.parent() {
fs::create_dir_all(parent)?;
}
write_entry_content(&dest_abs, entry)?;
if install_rel.ends_with(".zf.json") && install_rel.starts_with("pipelines/") {
let source = fs::read_to_string(&dest_abs)?;
let (title, trigger_kind) = infer_pipeline_meta(&source, &install_rel);
let meta = self.projects.upsert_pipeline_definition(
&target_owner,
&target_project,
&install_rel,
&title,
"",
&trigger_kind,
&source,
)?;
pipelines_registered.push(meta.file_rel_path);
}
}
Ok(HubInstallResult {
package_id: package_id.to_string(),
version: version.to_string(),
install_root,
files_written: payload.files.len(),
pipelines_registered,
})
}
fn review_artifact_payload(
&self,
target_owner: &str,
target_project: &str,
package_id: &str,
version: &str,
target_folder: &str,
payload: &HubArtifact,
) -> Result<HubInstallReview, PlatformError> {
let layout = self.projects.project_layout(target_owner, target_project)?;
let install_root =
install_root_for_target_folder(package_id, &payload.asset_kind, target_folder);
let mut files_added = Vec::new();
let mut files_overwritten = Vec::new();
let mut pipelines_registered = Vec::new();
let mut policy_entries = Vec::new();
for entry in &payload.files {
let install_rel = install_rel_path_under_folder(&install_root, &entry.rel_path);
let dest_abs = layout.repo_dir.join(&install_rel);
if dest_abs.exists() {
files_overwritten.push(install_rel.clone());
} else {
files_added.push(install_rel.clone());
}
if install_rel.ends_with(".zf.json") && install_rel.starts_with("pipelines/") {
pipelines_registered.push(install_rel.clone());
}
policy_entries.push(package_policy_entry(&install_rel, entry));
}
let mut policy =
review_package_entries(&policy_entries, Vec::new(), PackageReviewOptions::default());
policy.risk_level = install_risk_level(&policy, !files_overwritten.is_empty());
Ok(HubInstallReview {
package_id: package_id.to_string(),
version: version.to_string(),
target_folder: target_folder.to_string(),
install_root,
asset_kind: payload.asset_kind.clone(),
files_added,
files_overwritten,
pipelines_registered,
nodes_used: policy.nodes_used,
credentials_required: policy.credentials_required,
external_urls: policy.external_urls,
database_effects: policy.database_effects,
filesystem_effects: policy.filesystem_effects,
public_endpoints: policy.public_endpoints,
schedules: policy.schedules,
large_files: policy.large_files,
seed_data: policy.seed_data,
project_initialization: serde_json::to_value(&payload.project_initialization)
.unwrap_or_else(|_| Value::Null),
warnings: policy.warnings,
risk_level: policy.risk_level,
})
}
pub fn get_asset_version_artifact(
&self,
package_id: &str,
version: &str,
) -> Result<(HubAssetVersion, Value), PlatformError> {
let (version_row, artifact, _) =
self.get_asset_version_install_artifact(package_id, version)?;
Ok((version_row, artifact))
}
pub fn get_asset_version_install_artifact(
&self,
package_id: &str,
version: &str,
) -> Result<(HubAssetVersion, Value, u64), PlatformError> {
self.require_enabled()?;
let Some(version_row) = self.hub_data.get_hub_asset_version(package_id, version)? else {
return Err(PlatformError::new(
"HUB_ASSET_MISSING",
"asset version not found",
));
};
let artifact_abs = self.hub_artifact_path(&version_row.artifact_rel_path)?;
let artifact_size_bytes = fs::metadata(&artifact_abs)?.len();
if artifact_size_bytes > MAX_REMOTE_HUB_ARTIFACT_BYTES {
return Err(PlatformError::new(
"HUB_ARTIFACT_TOO_LARGE",
"hub artifact exceeds maximum read size",
));
}
let raw = fs::read_to_string(&artifact_abs)?;
let artifact = serde_json::from_str::<Value>(&raw)
.map_err(|err| PlatformError::new("HUB_INSTALL", err.to_string()))?;
Ok((version_row, artifact, artifact_size_bytes))
}
pub fn get_latest_asset_media(
&self,
package_id: &str,
media_name: &str,
) -> Result<(HubAssetPackage, HubMediaFile, Vec<u8>), PlatformError> {
self.require_enabled()?;
let Some(package) = self.hub_data.get_hub_asset_package(package_id)? else {
return Err(PlatformError::new(
"HUB_ASSET_MISSING",
"asset package not found",
));
};
let Some(version) = self
.hub_data
.list_hub_asset_versions(package_id)?
.into_iter()
.next()
else {
return Err(PlatformError::new(
"HUB_ASSET_MISSING",
"asset version not found",
));
};
let artifact_abs = self.hub_artifact_path(&version.artifact_rel_path)?;
let raw = fs::read_to_string(&artifact_abs)?;
let artifact: HubArtifact = serde_json::from_str(&raw)
.map_err(|err| PlatformError::new("HUB_INSTALL", err.to_string()))?;
let Some(media) = artifact
.media
.into_iter()
.find(|item| item.name == media_name)
else {
return Err(PlatformError::new("HUB_MEDIA_MISSING", "media not found"));
};
let bytes = base64::engine::general_purpose::STANDARD
.decode(&media.content)
.map_err(|err| PlatformError::new("HUB_MEDIA_INVALID", err.to_string()))?;
if sha256_hex(&bytes) != media.sha256 {
return Err(PlatformError::new(
"HUB_MEDIA_INVALID",
"media hash mismatch",
));
}
Ok((package, media, bytes))
}
pub fn import_remote_asset(
&self,
authority_owner: &str,
authority_project: &str,
token: &HubToken,
req: &RemoteHubPublishRequest,
) -> Result<(HubAssetPackage, HubAssetVersion), PlatformError> {
let _ = (authority_owner, authority_project);
self.require_enabled()?;
let authority_owner = HUB_SERVICE_SCOPE_OWNER.to_string();
let authority_project = HUB_SERVICE_SCOPE_PROJECT.to_string();
let publisher_owner = slug_segment(&token.owner);
let publisher_id = slug_segment(&token.publisher_id);
let package_id = canonical_hub_package_id(&publisher_id, &req.package_id)?;
let version = req.version.trim().to_string();
if authority_owner.is_empty()
|| authority_project.is_empty()
|| publisher_owner.is_empty()
|| publisher_id.is_empty()
|| package_id.is_empty()
|| version.is_empty()
{
return Err(PlatformError::new(
"HUB_REMOTE_INVALID",
"authority, publisher, package id, and version must not be empty",
));
}
validate_hub_version(&version)?;
let Some(publisher) = self.hub_data.get_hub_publisher(
HUB_SERVICE_SCOPE_OWNER,
HUB_SERVICE_SCOPE_PROJECT,
&publisher_id,
)?
else {
return Err(PlatformError::new(
"HUB_PUBLISHER_MISSING",
"publisher not found",
));
};
if !publisher.enabled {
return Err(PlatformError::new(
"HUB_PUBLISHER_DISABLED",
"publisher is disabled",
));
}
if !publisher.can_publish {
return Err(PlatformError::new(
"HUB_PUBLISHER_FORBIDDEN",
"publisher does not have publish permission",
));
}
let source_owner = slug_segment(&req.source_owner);
let source_project = slug_segment(&req.source_project);
let mut artifact: HubArtifact = serde_json::from_value(req.artifact.clone())
.map_err(|err| PlatformError::new("HUB_REMOTE_INVALID", err.to_string()))?;
if artifact.files.is_empty() {
return Err(PlatformError::new(
"HUB_REMOTE_INVALID",
"artifact must contain at least one file",
));
}
validate_hub_media(&artifact.media, &publisher)?;
validate_hub_gallery(&artifact.gallery, &artifact.media)?;
artifact.image_url = artifact
.gallery
.cover
.as_ref()
.map(|item| {
format!(
"/api/hub/remote/assets/{package_id}/media/{}",
item.media_name
)
})
.or_else(|| {
artifact
.media
.iter()
.find(|item| item.role == "cover")
.map(|item| format!("/api/hub/remote/assets/{package_id}/media/{}", item.name))
})
.unwrap_or_default();
sanitize_hub_export_entries(&mut artifact.files)?;
let artifact_value = serde_json::to_value(&artifact)
.map_err(|err| PlatformError::new("HUB_REMOTE_INVALID", err.to_string()))?;
let artifact_rel = format!(
"services/{}/packages/{}/versions/{}/artifact.json",
DEFAULT_HUB_SERVICE_INSTANCE_ID, package_id, version
);
let artifact_abs = self.data_root.join(&artifact_rel);
if let Some(parent) = artifact_abs.parent() {
fs::create_dir_all(parent)?;
}
let artifact_bytes = serde_json::to_vec_pretty(&artifact)
.map_err(|err| PlatformError::new("HUB_REMOTE_INVALID", err.to_string()))?;
let now = now_ts();
let authority = self.ensure_service_authority()?;
let existing_package = self.hub_data.get_hub_asset_package(&package_id)?;
self.enforce_publisher_package_quota(
&publisher,
&package_id,
existing_package.as_ref(),
artifact_bytes.len(),
)?;
fs::write(&artifact_abs, &artifact_bytes)?;
let package = HubAssetPackage {
package_pk: existing_package
.as_ref()
.map(|item| item.package_pk.clone())
.filter(|value| !value.is_empty())
.unwrap_or_else(|| format!("mpkg_{}", random_hex(8))),
authority_id: authority.authority_id.clone(),
publisher_pk: publisher.publisher_pk.clone(),
package_id: package_id.clone(),
authority_owner: authority_owner.clone(),
authority_project: authority_project.clone(),
publisher_owner: publisher_owner.clone(),
publisher_id: publisher_id.clone(),
publisher_display_name: publisher.display_name.clone(),
publisher_url: publisher.publisher_url.clone(),
publisher_email: publisher.email.clone(),
asset_kind: validate_hub_asset_kind(&artifact.asset_kind)?,
title: if req.title.trim().is_empty() {
artifact.title.clone()
} else {
req.title.trim().to_string()
},
description: if req.description.trim().is_empty() {
artifact.description.clone()
} else {
req.description.trim().to_string()
},
image_url: artifact.image_url.clone(),
visibility: normalize_visibility(&req.visibility),
tags: req.tags.clone(),
created_at: existing_package.map(|item| item.created_at).unwrap_or(now),
updated_at: now,
};
let version_row = HubAssetVersion {
package_pk: package.package_pk.clone(),
package_id,
version,
authority_owner,
authority_project,
publisher_owner,
publisher_id,
source_owner,
source_project,
source_kind: req.source_kind.clone(),
source_ref: req.source_ref.clone(),
artifact_rel_path: artifact_rel,
artifact_sha256: sha256_hex(&artifact_bytes),
manifest: artifact_value,
created_at: now,
};
self.hub_data.put_hub_asset_package(&package)?;
self.hub_data.put_hub_asset_version(&version_row)?;
Ok((package, version_row))
}
pub async fn fetch_remote_pack_rows(
&self,
http_client: &reqwest::Client,
owner: &str,
project: &str,
) -> Result<Vec<HubRemotePackRow>, PlatformError> {
let repos = self.list_effective_repositories(owner, project)?;
let mut out = Vec::new();
for repo in repos.into_iter().filter(|item| item.enabled) {
let url = remote_hub_url(&repo, "remote/assets");
validate_remote_hub_url(&url)?;
let mut req = http_client.get(url);
if !repo.read_token.trim().is_empty() {
req = req.bearer_auth(repo.read_token.trim());
}
let response = req
.send()
.await
.map_err(|err| PlatformError::new("HUB_REMOTE_FETCH", err.to_string()))?;
if !response.status().is_success() {
continue;
}
let payload: RemoteHubListResponse = response
.json()
.await
.map_err(|err| PlatformError::new("HUB_REMOTE_FETCH", err.to_string()))?;
out.extend(payload.items.into_iter().map(|item| HubRemotePackRow {
repository_id: repo.repository_id.clone(),
repository_title: repo.title.clone(),
package_id: item.package_id,
publisher_owner: item.publisher_owner,
publisher_id: item.publisher_id,
publisher_display_name: item.publisher_display_name,
publisher_url: item.publisher_url,
publisher_email: item.publisher_email,
asset_kind: item.asset_kind,
title: item.title,
description: item.description,
summary: item.summary,
image_url: item.image_url,
gallery: item.gallery,
visibility: item.visibility,
tags: item.tags,
latest_version: item.latest_version,
updated_at: item.updated_at,
source: "remote".to_string(),
}));
}
Ok(out)
}
pub async fn fetch_platform_remote_app_rows(
&self,
http_client: &reqwest::Client,
owner: &str,
) -> Result<Vec<HubRemotePackRow>, PlatformError> {
let repos = self.list_platform_repositories(owner)?;
self.fetch_platform_remote_app_rows_from_repositories(http_client, repos)
.await
}
pub async fn fetch_platform_remote_app_rows_from_repositories(
&self,
http_client: &reqwest::Client,
repos: Vec<PlatformHubRepository>,
) -> Result<Vec<HubRemotePackRow>, PlatformError> {
let mut out = Vec::new();
for repo in repos.into_iter().filter(|item| item.enabled) {
let url = remote_hub_url_for_platform(&repo, "remote/assets");
validate_remote_hub_url(&url)?;
let mut req = http_client.get(url);
if !repo.read_token.trim().is_empty() {
req = req.bearer_auth(repo.read_token.trim());
}
let response = match req.send().await {
Ok(response) => response,
Err(_) => continue,
};
if !response.status().is_success() {
continue;
}
let payload: RemoteHubListResponse = match response.json().await {
Ok(payload) => payload,
Err(_) => continue,
};
out.extend(
payload
.items
.into_iter()
.filter(|item| item.asset_kind == HUB_ASSET_KIND_PROJECT_BUNDLE)
.map(|item| HubRemotePackRow {
repository_id: repo.repository_id.clone(),
repository_title: repo.title.clone(),
package_id: item.package_id,
publisher_owner: item.publisher_owner,
publisher_id: item.publisher_id,
publisher_display_name: item.publisher_display_name,
publisher_url: item.publisher_url,
publisher_email: item.publisher_email,
asset_kind: item.asset_kind,
title: item.title,
description: item.description,
summary: item.summary,
image_url: item.image_url,
gallery: item.gallery,
visibility: item.visibility,
tags: item.tags,
latest_version: item.latest_version,
updated_at: item.updated_at,
source: "remote".to_string(),
}),
);
}
Ok(out)
}
pub async fn install_remote_pack_from_repository(
&self,
http_client: &reqwest::Client,
target_owner: &str,
target_project: &str,
repository_id: &str,
package_id: &str,
version: &str,
target_folder: &str,
) -> Result<HubInstallResult, PlatformError> {
let repo = self
.list_effective_repositories(target_owner, target_project)?
.into_iter()
.find(|item| item.repository_id == repository_id)
.ok_or_else(|| PlatformError::new("HUB_REPOSITORY_MISSING", "repository not found"))?;
let url = remote_hub_url(
&repo,
&format!("remote/assets/{}/{}/artifact", package_id, version),
);
validate_remote_hub_url(&url)?;
let mut req = http_client.get(url);
if !repo.read_token.trim().is_empty() {
req = req.bearer_auth(repo.read_token.trim());
}
let response = req
.send()
.await
.map_err(|err| PlatformError::new("HUB_REMOTE_FETCH", err.to_string()))?;
if !response.status().is_success() {
return Err(PlatformError::new(
"HUB_REMOTE_FETCH",
format!("remote fetch failed with {}", response.status()),
));
}
let payload: RemoteHubArtifactResponse = response
.json()
.await
.map_err(|err| PlatformError::new("HUB_REMOTE_FETCH", err.to_string()))?;
verify_remote_artifact_hash(&payload)?;
let artifact: HubArtifact = serde_json::from_value(payload.artifact)
.map_err(|err| PlatformError::new("HUB_REMOTE_INVALID", err.to_string()))?;
self.install_artifact_payload(
slug_segment(target_owner),
slug_segment(target_project),
package_id,
version,
target_folder,
artifact,
)
}
pub async fn review_remote_pack_from_repository(
&self,
http_client: &reqwest::Client,
target_owner: &str,
target_project: &str,
repository_id: &str,
package_id: &str,
version: &str,
target_folder: &str,
) -> Result<HubInstallReview, PlatformError> {
let repo = self
.list_effective_repositories(target_owner, target_project)?
.into_iter()
.find(|item| item.repository_id == repository_id)
.ok_or_else(|| PlatformError::new("HUB_REPOSITORY_MISSING", "repository not found"))?;
let url = remote_hub_url(
&repo,
&format!("remote/assets/{}/{}/artifact", package_id, version),
);
validate_remote_hub_url(&url)?;
let mut req = http_client.get(url);
if !repo.read_token.trim().is_empty() {
req = req.bearer_auth(repo.read_token.trim());
}
let response = req
.send()
.await
.map_err(|err| PlatformError::new("HUB_REMOTE_FETCH", err.to_string()))?;
if !response.status().is_success() {
return Err(PlatformError::new(
"HUB_REMOTE_FETCH",
format!("remote fetch failed with {}", response.status()),
));
}
let payload: RemoteHubArtifactResponse = response
.json()
.await
.map_err(|err| PlatformError::new("HUB_REMOTE_FETCH", err.to_string()))?;
verify_remote_artifact_hash(&payload)?;
let artifact: HubArtifact = serde_json::from_value(payload.artifact)
.map_err(|err| PlatformError::new("HUB_REMOTE_INVALID", err.to_string()))?;
self.review_artifact_payload(
&slug_segment(target_owner),
&slug_segment(target_project),
package_id,
version,
target_folder,
&artifact,
)
}
pub async fn install_remote_project_from_platform_repository(
&self,
http_client: &reqwest::Client,
owner: &str,
repository_id: &str,
package_id: &str,
version: &str,
) -> Result<(String, String), PlatformError> {
self.install_remote_project_from_platform_source(
http_client,
owner,
owner,
repository_id,
package_id,
version,
)
.await
}
pub async fn install_remote_project_from_platform_source(
&self,
http_client: &reqwest::Client,
source_owner: &str,
target_owner: &str,
repository_id: &str,
package_id: &str,
version: &str,
) -> Result<(String, String), PlatformError> {
let source_owner = slug_segment(source_owner);
let target_owner = slug_segment(target_owner);
let repo = self
.list_platform_repositories(&source_owner)?
.into_iter()
.find(|item| item.repository_id == repository_id)
.ok_or_else(|| PlatformError::new("HUB_REPOSITORY_MISSING", "repository not found"))?;
let url = remote_hub_url_for_platform(
&repo,
&format!("remote/assets/{}/{}/artifact", package_id, version),
);
validate_remote_hub_url(&url)?;
let mut req = http_client.get(url);
if !repo.read_token.trim().is_empty() {
req = req.bearer_auth(repo.read_token.trim());
}
let response = req
.send()
.await
.map_err(|err| PlatformError::new("HUB_REMOTE_FETCH", err.to_string()))?;
if !response.status().is_success() {
return Err(PlatformError::new(
"HUB_REMOTE_FETCH",
format!("remote fetch failed with {}", response.status()),
));
}
let payload: RemoteHubArtifactResponse = response
.json()
.await
.map_err(|err| PlatformError::new("HUB_REMOTE_FETCH", err.to_string()))?;
verify_remote_artifact_hash(&payload)?;
let artifact: HubArtifact = serde_json::from_value(payload.artifact.clone())
.map_err(|err| PlatformError::new("HUB_REMOTE_INVALID", err.to_string()))?;
if validate_hub_asset_kind(&artifact.asset_kind)? != HUB_ASSET_KIND_PROJECT_BUNDLE {
return Err(PlatformError::new(
"HUB_REMOTE_INVALID",
"platform hub install only supports project bundles",
));
}
let base_project = slug_segment(package_id);
if base_project.is_empty() {
return Err(PlatformError::new(
"HUB_REMOTE_INVALID",
"package id is not a valid project slug",
));
}
let project_title = if artifact.title.trim().is_empty() {
package_id.to_string()
} else {
artifact.title.clone()
};
let mut suffix = 1usize;
let project = loop {
let candidate = if suffix == 1 {
base_project.clone()
} else {
format!("{base_project}-{suffix}")
};
if self
.control_data
.get_project(&target_owner, &candidate)?
.is_some()
{
suffix += 1;
continue;
}
match self.projects.create_or_update_project(
&target_owner,
&CreateProjectRequest {
project: candidate.clone(),
title: Some(project_title.clone()),
local_branch: None,
runtime: ProjectRuntimeSelectionRequest::default(),
},
) {
Ok(_) => break candidate,
Err(err)
if err.code == "PLATFORM_GIT_INIT"
|| err.code == "PROJECT_EXISTS"
|| err.code == "PLATFORM_PROJECT_EXISTS" =>
{
suffix += 1;
continue;
}
Err(err) => return Err(err),
}
};
let layout = self.projects.project_layout(&target_owner, &project)?;
clear_repo_worktree_preserving_git(&layout.repo_dir)?;
for entry in &artifact.files {
let dest_abs = sanitize_install_repo_path(&layout, &entry.rel_path)?;
if let Some(parent) = dest_abs.parent() {
fs::create_dir_all(parent)?;
}
write_entry_content(&dest_abs, entry)?;
}
reindex_project_bundle_pipelines(self, &target_owner, &project, &artifact.files)?;
sekejap::apply_schema_from_repo(&self.data_root, &target_owner, &project)?;
sqlite_schema::apply_schema_from_repo(&self.data_root, &target_owner, &project)?;
execute_project_initial_data(
&self.data_root,
&target_owner,
&project,
&layout,
&artifact.project_initialization.initial_data,
)?;
for file_rel_path in &artifact.active_pipelines {
let normalized = normalize_repo_rel(file_rel_path);
if normalized.is_empty() || !normalized.ends_with(".zf.json") {
continue;
}
self.projects
.activate_pipeline_definition(&target_owner, &project, &normalized)?;
}
Ok((target_owner, project))
}
fn preview_pipeline(
&self,
owner: &str,
project: &str,
layout: &ProjectFileLayout,
source_ref: &str,
) -> Result<HubExportPreview, PlatformError> {
let Some(meta) = self
.projects
.get_pipeline_meta_by_file_id(owner, project, source_ref)?
else {
return Err(PlatformError::new(
"HUB_PUBLISH_MISSING",
format!("pipeline '{}' not found", source_ref),
));
};
let mut warnings = Vec::new();
let mut entries = vec![read_repo_entry(
layout,
&meta.file_rel_path,
"primary pipeline".to_string(),
)?];
let source = self
.projects
.read_pipeline_source(owner, project, &meta.file_rel_path)?;
let value: Value = serde_json::from_str(&source)
.map_err(|err| PlatformError::new("HUB_PREVIEW", err.to_string()))?;
let mut seen = BTreeSet::new();
seen.insert(meta.file_rel_path.clone());
if let Some(nodes) = value.get("nodes").and_then(Value::as_array) {
for node in nodes {
if node.get("kind").and_then(Value::as_str) != Some("n.web.response") {
continue;
}
let Some(template_rel) = node
.get("config")
.and_then(|cfg| cfg.get("template"))
.and_then(Value::as_str)
else {
continue;
};
self.collect_template_dependency_entries(
layout,
template_rel,
"web response template".to_string(),
&mut seen,
&mut entries,
&mut warnings,
)?;
}
}
entries.sort_by(|a, b| a.rel_path.cmp(&b.rel_path));
Ok(build_preview(
HUB_ASSET_KIND_PIPELINE_BUNDLE.to_string(),
"pipeline_with_dependencies".to_string(),
meta.file_rel_path.clone(),
meta.title,
if meta.description.trim().is_empty() {
"Pipeline export with local project dependencies".to_string()
} else {
meta.description
},
entries,
warnings,
))
}
fn preview_template(
&self,
owner: &str,
project: &str,
layout: &ProjectFileLayout,
source_ref: &str,
) -> Result<HubExportPreview, PlatformError> {
let listing = self.projects.list_template_workspace(owner, project)?;
let selected = listing
.items
.into_iter()
.find(|item| item.kind != "folder" && item.rel_path == source_ref)
.ok_or_else(|| {
PlatformError::new(
"HUB_PUBLISH_MISSING",
format!("template '{}' not found", source_ref),
)
})?;
let mut warnings = Vec::new();
let mut entries = Vec::new();
let mut seen = BTreeSet::new();
self.collect_template_dependency_entries(
layout,
&selected.rel_path,
"primary template".to_string(),
&mut seen,
&mut entries,
&mut warnings,
)?;
entries.sort_by(|a, b| a.rel_path.cmp(&b.rel_path));
Ok(build_preview(
HUB_ASSET_KIND_TEMPLATE_BUNDLE.to_string(),
"template_with_dependencies".to_string(),
selected.rel_path.clone(),
selected.name,
format!(
"Template export with {} local dependencies",
entries.len().saturating_sub(1)
),
entries,
warnings,
))
}
fn preview_folder(
&self,
layout: &ProjectFileLayout,
source_ref: &str,
) -> Result<HubExportPreview, PlatformError> {
let mut entries = collect_tree_entries(layout, source_ref)?;
entries.sort_by(|a, b| a.rel_path.cmp(&b.rel_path));
let name = Path::new(source_ref)
.file_name()
.and_then(|v| v.to_str())
.unwrap_or(source_ref)
.to_string();
Ok(build_preview(
HUB_ASSET_KIND_FOLDER_BUNDLE.to_string(),
"folder_files".to_string(),
source_ref.to_string(),
name,
"Recursive folder export".to_string(),
entries,
Vec::new(),
))
}
fn preview_project(
&self,
layout: &ProjectFileLayout,
) -> Result<HubExportPreview, PlatformError> {
let mut entries = collect_tree_entries(layout, ".")?;
entries.sort_by(|a, b| a.rel_path.cmp(&b.rel_path));
Ok(build_preview(
HUB_ASSET_KIND_PROJECT_BUNDLE.to_string(),
"project_files".to_string(),
".".to_string(),
"Project files".to_string(),
"Full repo export".to_string(),
entries,
Vec::new(),
))
}
fn collect_template_dependency_entries(
&self,
layout: &ProjectFileLayout,
rel_path: &str,
reason: String,
seen: &mut BTreeSet<String>,
entries: &mut Vec<HubExportEntry>,
warnings: &mut Vec<String>,
) -> Result<(), PlatformError> {
let normalized = normalize_template_repo_rel(rel_path);
if normalized.is_empty() || !seen.insert(normalized.clone()) {
return Ok(());
}
let abs = layout.repo_dir.join(&normalized);
if !abs.is_file() {
warnings.push(format!("Skipped missing dependency '{}'", normalized));
return Ok(());
}
let entry = read_repo_entry(layout, &normalized, reason)?;
let source_text = if entry.encoding == "text" {
Some(entry.content.clone())
} else {
None
};
entries.push(entry);
let Some(source_text) = source_text else {
return Ok(());
};
let import_sources = extract_import_sources(&source_text);
for src in import_sources {
match resolve_local_import(layout, &normalized, &src) {
Some(dep_rel) => self.collect_template_dependency_entries(
layout,
&dep_rel,
format!("imported from {}", normalized),
seen,
entries,
warnings,
)?,
None => {
if src.starts_with("@/") || src.starts_with('.') {
warnings.push(format!(
"Skipped unresolved local import '{}' from '{}'",
src, normalized
));
}
}
}
}
Ok(())
}
}
impl HubService {
fn require_enabled(&self) -> Result<PlatformServiceInstance, PlatformError> {
match self.get_default_service_instance()? {
Some(service) if service.enabled => Ok(service),
_ => Err(PlatformError::new(
"HUB_SERVICE_DISABLED",
"hub service is not enabled",
)),
}
}
fn ensure_service_authority(&self) -> Result<HubAuthority, PlatformError> {
if let Some(authority) = self
.hub_data
.get_hub_authority(HUB_SERVICE_SCOPE_OWNER, HUB_SERVICE_SCOPE_PROJECT)?
{
return Ok(authority);
}
let service = self.require_enabled()?;
let now = now_ts();
let authority = HubAuthority {
authority_id: DEFAULT_HUB_SERVICE_INSTANCE_ID.to_string(),
host_project_id: service.host_office_id,
owner: HUB_SERVICE_SCOPE_OWNER.to_string(),
project: HUB_SERVICE_SCOPE_PROJECT.to_string(),
enabled: true,
public_base_url: service.public_base_url,
created_at: now,
updated_at: now,
};
self.hub_data.put_hub_authority(&authority)?;
Ok(authority)
}
fn hub_artifact_path(&self, rel_path: &str) -> Result<PathBuf, PlatformError> {
let rel_path = rel_path.trim().replace('\\', "/");
if rel_path.is_empty() || rel_path.contains('\0') {
return Err(PlatformError::new(
"HUB_ARTIFACT_PATH_INVALID",
"artifact path is invalid",
));
}
let rel = Path::new(&rel_path);
if rel.is_absolute()
|| rel
.components()
.any(|component| !matches!(component, std::path::Component::Normal(_)))
{
return Err(PlatformError::new(
"HUB_ARTIFACT_PATH_INVALID",
"artifact path must be a contained relative path",
));
}
let service_rel_prefix = format!("services/{DEFAULT_HUB_SERVICE_INSTANCE_ID}/");
if !rel_path.starts_with(&service_rel_prefix) {
return Err(PlatformError::new(
"HUB_ARTIFACT_PATH_INVALID",
"artifact path must live under the hub service root",
));
}
let service_root = self
.data_root
.join("services")
.join(DEFAULT_HUB_SERVICE_INSTANCE_ID);
let abs = self.data_root.join(rel);
let service_root = fs::canonicalize(&service_root)?;
let abs = fs::canonicalize(&abs)?;
if !abs.starts_with(&service_root) {
return Err(PlatformError::new(
"HUB_ARTIFACT_PATH_INVALID",
"artifact path escapes hub service root",
));
}
Ok(abs)
}
fn hub_artifact_path_for_delete(&self, rel_path: &str) -> Result<PathBuf, PlatformError> {
let rel_path = rel_path.trim().replace('\\', "/");
if rel_path.is_empty() || rel_path.contains('\0') {
return Err(PlatformError::new(
"HUB_ARTIFACT_PATH_INVALID",
"artifact path is invalid",
));
}
let rel = Path::new(&rel_path);
if rel.is_absolute()
|| rel
.components()
.any(|component| !matches!(component, std::path::Component::Normal(_)))
{
return Err(PlatformError::new(
"HUB_ARTIFACT_PATH_INVALID",
"artifact path must be a contained relative path",
));
}
let service_rel_prefix = format!("services/{DEFAULT_HUB_SERVICE_INSTANCE_ID}/");
if !rel_path.starts_with(&service_rel_prefix) {
return Err(PlatformError::new(
"HUB_ARTIFACT_PATH_INVALID",
"artifact path must live under the hub service root",
));
}
Ok(self.data_root.join(rel))
}
fn enforce_publisher_package_quota(
&self,
publisher: &HubPublisher,
package_id: &str,
existing_package: Option<&HubAssetPackage>,
artifact_bytes: usize,
) -> Result<(), PlatformError> {
let max_package_bytes = normalize_limit(
publisher.max_package_bytes,
DEFAULT_PUBLISHER_MAX_PACKAGE_BYTES,
);
let max_packages = normalize_limit(publisher.max_packages, DEFAULT_PUBLISHER_MAX_PACKAGES);
if artifact_bytes as i64 > max_package_bytes {
return Err(PlatformError::new(
"HUB_PUBLISHER_QUOTA_EXCEEDED",
"publisher artifact byte limit exceeded",
));
}
if let Some(existing) = existing_package {
if existing.publisher_id != publisher.publisher_id
|| existing.publisher_pk != publisher.publisher_pk
{
return Err(PlatformError::new(
"HUB_PACKAGE_FORBIDDEN",
"package id already belongs to another publisher",
));
}
return Ok(());
}
let package_count = self
.hub_data
.list_hub_asset_packages()?
.into_iter()
.filter(|item| {
item.publisher_id == publisher.publisher_id
&& item.publisher_pk == publisher.publisher_pk
})
.count() as i64;
if package_count >= max_packages {
return Err(PlatformError::new(
"HUB_PUBLISHER_QUOTA_EXCEEDED",
"publisher package count limit exceeded",
));
}
if package_id.is_empty() {
return Err(PlatformError::new(
"HUB_PUBLISH_INVALID",
"package id must not be empty",
));
}
Ok(())
}
}
fn remote_hub_api_base(base_url: &str) -> String {
base_url.trim().trim_end_matches('/').to_string()
}
fn is_direct_hub_base(base_url: &str) -> bool {
let base = remote_hub_api_base(base_url).to_lowercase();
is_legacy_project_hub_base(&base) || is_ownerless_hub_base(&base)
}
fn is_legacy_project_hub_base(base_url: &str) -> bool {
let base = remote_hub_api_base(base_url).to_lowercase();
base.contains("/api/projects/") && base.ends_with("/hub")
}
fn is_ownerless_hub_base(base_url: &str) -> bool {
let base = remote_hub_api_base(base_url).to_lowercase();
base.ends_with("/api") || base.ends_with("/api/hub")
}
fn ownerless_hub_url(api_base: &str, suffix: &str) -> String {
if api_base.to_lowercase().ends_with("/api/hub") {
format!("{}/{}", api_base, suffix)
} else {
format!("{}/hub/{}", api_base, suffix)
}
}
fn remote_hub_url(repo: &ProjectHubRepository, suffix: &str) -> String {
let api_base = remote_hub_api_base(&repo.base_url);
let suffix = suffix.trim_start_matches('/');
if is_legacy_project_hub_base(&api_base) {
format!("{}/{}", api_base, suffix)
} else if is_ownerless_hub_base(&api_base) {
ownerless_hub_url(&api_base, suffix)
} else {
format!(
"{}/projects/{}/{}/hub/{}",
api_base, repo.remote_owner, repo.remote_project, suffix
)
}
}
fn remote_hub_url_for_platform(repo: &PlatformHubRepository, suffix: &str) -> String {
let api_base = remote_hub_api_base(&repo.base_url);
let suffix = suffix.trim_start_matches('/');
if is_legacy_project_hub_base(&api_base) {
format!("{}/{}", api_base, suffix)
} else if is_ownerless_hub_base(&api_base) {
ownerless_hub_url(&api_base, suffix)
} else {
format!(
"{}/projects/{}/{}/hub/{}",
api_base, repo.remote_owner, repo.remote_project, suffix
)
}
}
fn validate_remote_hub_url(url: &str) -> Result<(), PlatformError> {
if hub_localhost_remote_allowed(url) {
return Ok(());
}
crate::pipeline::security::validate_outbound_http_url(url, "hub.remote")
.map_err(|err| PlatformError::new(err.code, err.message))
}
fn hub_localhost_remote_allowed(url: &str) -> bool {
if std::env::var("ZEBFLOW_HUB_ALLOW_LOCALHOST_REMOTE")
.ok()
.as_deref()
!= Some("1")
{
return false;
}
let Ok(parsed) = reqwest::Url::parse(url) else {
return false;
};
matches!(
parsed.host_str().unwrap_or(""),
"localhost" | "127.0.0.1" | "::1"
)
}
fn clear_repo_worktree_preserving_git(repo_dir: &Path) -> Result<(), PlatformError> {
let entries =
fs::read_dir(repo_dir).map_err(|err| PlatformError::new("HUB_INSTALL", err.to_string()))?;
for entry in entries.flatten() {
let path = entry.path();
let keep = path
.file_name()
.and_then(|value| value.to_str())
.map(|name| name == ".git")
.unwrap_or(false);
if keep {
continue;
}
if path.is_dir() {
fs::remove_dir_all(&path)
.map_err(|err| PlatformError::new("HUB_INSTALL", err.to_string()))?;
} else {
fs::remove_file(&path)
.map_err(|err| PlatformError::new("HUB_INSTALL", err.to_string()))?;
}
}
Ok(())
}
fn sanitize_install_repo_path(
layout: &ProjectFileLayout,
rel_path: &str,
) -> Result<PathBuf, PlatformError> {
let cleaned = rel_path.trim_start_matches("./").trim_start_matches('/');
let abs = layout.repo_dir.join(cleaned);
if !abs.starts_with(&layout.repo_dir) {
return Err(PlatformError::new(
"HUB_INSTALL",
"artifact path escaped repo root",
));
}
Ok(abs)
}
fn build_preview(
asset_kind: String,
source_type: String,
source_ref: String,
name: String,
description: String,
entries: Vec<HubExportEntry>,
warnings: Vec<String>,
) -> HubExportPreview {
let total_bytes = entries.iter().map(|item| item.size_bytes).sum();
let total_files = entries.len();
HubExportPreview {
asset_kind,
source_type,
source_ref,
name,
description,
entries,
warnings,
total_files,
total_bytes,
}
}
fn text_export_entry(rel_path: &str, kind: &str, reason: &str, content: String) -> HubExportEntry {
HubExportEntry {
rel_path: rel_path.to_string(),
kind: kind.to_string(),
size_bytes: content.len(),
reason: reason.to_string(),
encoding: "text".to_string(),
content,
}
}
fn rewrite_project_libraries(
preview: &mut HubExportPreview,
selected_libraries: &[String],
) -> Result<(), PlatformError> {
let Some(entry) = preview
.entries
.iter_mut()
.find(|entry| normalize_repo_rel(&entry.rel_path) == "zebflow.json")
else {
return Ok(());
};
let Some(raw) = entry_text(entry) else {
return Ok(());
};
let Ok(mut cfg) = serde_json::from_str::<ZebflowJson>(&raw) else {
return Ok(());
};
let selected = selected_libraries.iter().cloned().collect::<BTreeSet<_>>();
cfg.configs
.rwe
.libraries
.retain(|name, _| selected.contains(name));
let content = serde_json::to_string_pretty(&cfg)
.map_err(|err| PlatformError::new("HUB_PUBLISH", err.to_string()))?
+ "\n";
entry.size_bytes = content.len();
entry.encoding = "text".to_string();
entry.content = content;
Ok(())
}
fn collect_publish_media_from_files(
layout: &ProjectFileLayout,
image_file_path: &str,
publisher: &HubPublisher,
) -> Result<Vec<HubMediaFile>, PlatformError> {
let image_file_path = image_file_path.trim();
if image_file_path.is_empty() {
return Ok(Vec::new());
}
let rel = normalize_object_path(image_file_path)
.map_err(|err| PlatformError::new("HUB_MEDIA_INVALID", err.to_string()))?;
let zebfs = LocalZebFs::new(layout.files_dir.clone());
let object = zebfs
.get(&rel)
.map_err(|err| PlatformError::new("HUB_MEDIA_MISSING", err.to_string()))?;
image_content_type_from_path(&rel)?;
let max_image_bytes =
normalize_limit(publisher.max_image_bytes, DEFAULT_PUBLISHER_MAX_IMAGE_BYTES) as usize;
let webp_bytes = normalize_hub_cover_to_webp(&object.bytes)?;
if webp_bytes.len() > max_image_bytes {
return Err(PlatformError::new(
"HUB_PUBLISHER_QUOTA_EXCEEDED",
"webp cover image exceeds publisher max image bytes",
));
}
let media_name = "cover.webp".to_string();
let sha256 = sha256_hex(&webp_bytes);
Ok(vec![HubMediaFile {
name: media_name,
role: "cover".to_string(),
content_type: "image/webp".to_string(),
size_bytes: webp_bytes.len(),
sha256,
encoding: "base64".to_string(),
content: base64::engine::general_purpose::STANDARD.encode(webp_bytes),
}])
}
fn validate_hub_media(
media: &[HubMediaFile],
publisher: &HubPublisher,
) -> Result<(), PlatformError> {
let max_media_files =
normalize_limit(publisher.max_media_files, DEFAULT_PUBLISHER_MAX_MEDIA_FILES) as usize;
if media.len() > max_media_files {
return Err(PlatformError::new(
"HUB_PUBLISHER_QUOTA_EXCEEDED",
"media file count exceeds publisher limit",
));
}
let max_image_bytes =
normalize_limit(publisher.max_image_bytes, DEFAULT_PUBLISHER_MAX_IMAGE_BYTES) as usize;
for item in media {
let normalized_name = normalize_media_name(&item.name)?;
if normalized_name != item.name {
return Err(PlatformError::new(
"HUB_MEDIA_INVALID",
"media name is not normalized",
));
}
if item.encoding != "base64" {
return Err(PlatformError::new(
"HUB_MEDIA_INVALID",
"media encoding must be base64",
));
}
validate_image_content_type(&item.content_type)?;
if item.size_bytes > max_image_bytes {
return Err(PlatformError::new(
"HUB_PUBLISHER_QUOTA_EXCEEDED",
"image exceeds publisher max image bytes",
));
}
let bytes = base64::engine::general_purpose::STANDARD
.decode(&item.content)
.map_err(|err| PlatformError::new("HUB_MEDIA_INVALID", err.to_string()))?;
if bytes.len() != item.size_bytes || sha256_hex(&bytes) != item.sha256 {
return Err(PlatformError::new(
"HUB_MEDIA_INVALID",
"media size or hash does not match content",
));
}
}
Ok(())
}
fn validate_hub_gallery(gallery: &HubGallery, media: &[HubMediaFile]) -> Result<(), PlatformError> {
let media_names = media
.iter()
.map(|item| item.name.as_str())
.collect::<BTreeSet<_>>();
if let Some(cover) = &gallery.cover {
if cover.kind.trim() != "image" {
return Err(PlatformError::new(
"HUB_GALLERY_INVALID",
"gallery cover kind must be image",
));
}
validate_gallery_media_ref(&cover.media_name, &media_names)?;
}
for item in &gallery.items {
match item.kind.trim() {
"image" => validate_gallery_media_ref(&item.media_name, &media_names)?,
"youtube" => validate_youtube_url(&item.url)?,
_ => {
return Err(PlatformError::new(
"HUB_GALLERY_INVALID",
"gallery item kind must be image or youtube",
));
}
}
}
Ok(())
}
fn review_publish_artifact(
package_id: String,
version: String,
preview: HubExportPreview,
title: String,
description: String,
visibility: String,
tags: Vec<String>,
media: Vec<HubMediaFile>,
project_initialization: HubProjectInitialization,
) -> Result<HubPublishReview, PlatformError> {
let policy_entries = preview
.entries
.iter()
.map(|entry| package_policy_entry(&entry.rel_path, entry))
.collect::<Vec<_>>();
let policy = review_package_entries(
&policy_entries,
preview.warnings.clone(),
PackageReviewOptions {
publish_mode: true,
require_title: true,
require_description: true,
require_cover_image: true,
title: title.clone(),
description: description.clone(),
has_cover_image: !media.is_empty(),
},
);
Ok(HubPublishReview {
package_id,
version,
asset_kind: preview.asset_kind.clone(),
source_type: preview.source_type.clone(),
source_ref: preview.source_ref.clone(),
title,
description,
visibility,
tags,
total_files: preview.total_files,
total_bytes: preview.total_bytes,
files: preview.entries.clone(),
media: media
.iter()
.map(|item| HubPublishMediaReview {
name: item.name.clone(),
role: item.role.clone(),
content_type: item.content_type.clone(),
size_bytes: item.size_bytes,
})
.collect(),
nodes_used: policy.nodes_used,
credentials_required: policy.credentials_required,
external_urls: policy.external_urls,
database_effects: policy.database_effects,
filesystem_effects: policy.filesystem_effects,
public_endpoints: policy.public_endpoints,
schedules: policy.schedules,
large_files: policy.large_files,
seed_data: policy.seed_data,
project_initialization: serde_json::to_value(project_initialization)
.map_err(|err| PlatformError::new("HUB_PUBLISH_REVIEW", err.to_string()))?,
warnings: policy.warnings,
risk_level: policy.risk_level,
})
}
fn normalize_hub_cover_to_webp(bytes: &[u8]) -> Result<Vec<u8>, PlatformError> {
const MAX_DIM: u32 = 8000;
const MAX_DECODED_BYTES: u64 = 128 * 1024 * 1024;
let (w, h) = image::ImageReader::new(Cursor::new(bytes))
.with_guessed_format()
.map_err(|err| PlatformError::new("HUB_MEDIA_INVALID", err.to_string()))?
.into_dimensions()
.map_err(|err| PlatformError::new("HUB_MEDIA_INVALID", err.to_string()))?;
if w == 0 || h == 0 || w > MAX_DIM || h > MAX_DIM {
return Err(PlatformError::new(
"HUB_MEDIA_INVALID",
"cover image dimensions are invalid or too large",
));
}
if (w as u64) * (h as u64) * 4 > MAX_DECODED_BYTES {
return Err(PlatformError::new(
"HUB_MEDIA_INVALID",
"cover image decoded size is too large",
));
}
let image = image::ImageReader::new(Cursor::new(bytes))
.with_guessed_format()
.map_err(|err| PlatformError::new("HUB_MEDIA_INVALID", err.to_string()))?
.decode()
.map_err(|err| PlatformError::new("HUB_MEDIA_INVALID", err.to_string()))?;
let image = fit_hub_cover(image, 1200, 675);
let mut out = Vec::new();
image
.write_to(&mut Cursor::new(&mut out), ImageFormat::WebP)
.map_err(|err| PlatformError::new("HUB_MEDIA_INVALID", err.to_string()))?;
Ok(out)
}
fn fit_hub_cover(img: DynamicImage, max_w: u32, max_h: u32) -> DynamicImage {
let (w, h) = img.dimensions();
if w <= max_w && h <= max_h {
return img;
}
img.resize(max_w, max_h, FilterType::Lanczos3)
}
fn validate_gallery_media_ref(
media_name: &str,
media_names: &BTreeSet<&str>,
) -> Result<(), PlatformError> {
let normalized = normalize_media_name(media_name)?;
if normalized != media_name {
return Err(PlatformError::new(
"HUB_GALLERY_INVALID",
"gallery media name is not normalized",
));
}
if !media_names.contains(media_name) {
return Err(PlatformError::new(
"HUB_GALLERY_INVALID",
"gallery image must reference an existing media item",
));
}
Ok(())
}
fn validate_youtube_url(input: &str) -> Result<(), PlatformError> {
let url = reqwest::Url::parse(input.trim()).map_err(|_| {
PlatformError::new("HUB_GALLERY_INVALID", "youtube URL must be a valid URL")
})?;
if url.scheme() != "https" {
return Err(PlatformError::new(
"HUB_GALLERY_INVALID",
"youtube URL must use https",
));
}
let host = url.host_str().unwrap_or_default().to_ascii_lowercase();
let ok_host = matches!(
host.as_str(),
"youtube.com" | "www.youtube.com" | "youtu.be" | "m.youtube.com"
);
if !ok_host {
return Err(PlatformError::new(
"HUB_GALLERY_INVALID",
"gallery video URL must be a YouTube URL",
));
}
Ok(())
}
fn normalize_media_name(input: &str) -> Result<String, PlatformError> {
let value = input.trim();
if value.is_empty()
|| value.contains('/')
|| value.contains('\\')
|| value.contains('\0')
|| value == "."
|| value == ".."
{
return Err(PlatformError::new(
"HUB_MEDIA_INVALID",
"media name must be a plain file name",
));
}
Ok(value.to_string())
}
fn image_content_type_from_path(path: &str) -> Result<String, PlatformError> {
let ext = Path::new(path)
.extension()
.and_then(|value| value.to_str())
.unwrap_or_default()
.to_ascii_lowercase();
let content_type = match ext.as_str() {
"png" => "image/png",
"jpg" | "jpeg" => "image/jpeg",
"webp" => "image/webp",
"gif" => "image/gif",
_ => {
return Err(PlatformError::new(
"HUB_MEDIA_INVALID",
"package image must be png, jpg, webp, or gif",
));
}
};
Ok(content_type.to_string())
}
fn normalize_hub_id_segment(raw: &str, label: &str) -> Result<String, PlatformError> {
let mut out = String::new();
let mut last_dash = false;
for c in raw.trim().to_ascii_lowercase().chars() {
if c.is_ascii_alphanumeric() {
out.push(c);
last_dash = false;
} else if c == '-' || c == '_' || c.is_ascii_whitespace() {
if !last_dash && !out.is_empty() {
out.push('-');
last_dash = true;
}
} else if !last_dash && !out.is_empty() {
out.push('-');
last_dash = true;
}
}
let value = out.trim_matches('-').to_string();
if value.is_empty() {
return Err(PlatformError::new(
"HUB_ID_INVALID",
format!("{label} must contain at least one alphanumeric character"),
));
}
Ok(value)
}
fn canonical_hub_package_id(
publisher_id: &str,
raw_package_id: &str,
) -> Result<String, PlatformError> {
let publisher = normalize_hub_id_segment(publisher_id, "publisher id")?;
let trimmed = raw_package_id.trim();
let asset_raw = if let Some((prefix, rest)) = trimmed.split_once('.') {
let normalized_prefix = normalize_hub_id_segment(prefix, "package publisher prefix")?;
if normalized_prefix != publisher {
return Err(PlatformError::new(
"HUB_PACKAGE_ID_INVALID",
"package id must use the selected publisher prefix",
));
}
rest
} else {
trimmed
};
let asset = normalize_hub_id_segment(asset_raw, "asset slug")?;
Ok(format!("{publisher}.{asset}"))
}
fn validate_image_content_type(content_type: &str) -> Result<(), PlatformError> {
match content_type.trim().to_ascii_lowercase().as_str() {
"image/png" | "image/jpeg" | "image/webp" | "image/gif" => Ok(()),
_ => Err(PlatformError::new(
"HUB_MEDIA_INVALID",
"media content type is not an allowed image type",
)),
}
}
fn normalize_visibility(input: &str) -> String {
match input.trim().to_ascii_lowercase().as_str() {
"public" => "public".to_string(),
"unlisted" => "unlisted".to_string(),
_ => "private".to_string(),
}
}
fn normalize_scopes(input: &[String]) -> Vec<String> {
let mut out = input
.iter()
.map(|item| item.trim().to_ascii_lowercase())
.filter(|item| matches!(item.as_str(), "hub:read" | "hub:publish" | "hub:manage"))
.collect::<Vec<_>>();
out.sort();
out.dedup();
out
}
fn validate_publisher_scopes(
publisher: &HubPublisher,
scopes: &[String],
) -> Result<(), PlatformError> {
for scope in scopes {
let allowed = match scope.as_str() {
"hub:read" => publisher.can_read,
"hub:publish" => publisher.can_publish,
"hub:manage" => publisher.can_manage,
_ => false,
};
if !allowed {
return Err(PlatformError::new(
"HUB_PUBLISHER_SCOPE_DENIED",
"publisher is not allowed to create a token with the requested scope",
));
}
}
Ok(())
}
fn apply_scope_flags(mut token: HubToken) -> HubToken {
token.scope_read = token.scopes.iter().any(|scope| scope == "hub:read");
token.scope_publish = token.scopes.iter().any(|scope| scope == "hub:publish");
token.scope_manage = token.scopes.iter().any(|scope| scope == "hub:manage");
token
}
fn normalize_publisher_url(publisher_id: &str, explicit: &str) -> String {
let trimmed = explicit.trim();
if !trimmed.is_empty() {
trimmed.to_string()
} else if publisher_id.is_empty() {
String::new()
} else {
format!("/publishers/{publisher_id}")
}
}
fn normalize_source_type(input: &str) -> String {
match input.trim() {
"template_with_dependencies" => "template_with_dependencies".to_string(),
"folder_files" => "folder_files".to_string(),
"project_files" => "project_files".to_string(),
_ => "pipeline_with_dependencies".to_string(),
}
}
fn validate_hub_asset_kind(input: &str) -> Result<String, PlatformError> {
let value = input.trim();
if HUB_ASSET_KINDS.contains(&value) {
Ok(value.to_string())
} else {
Err(PlatformError::new(
"HUB_ASSET_KIND_INVALID",
format!("asset kind must be one of: {}", HUB_ASSET_KINDS.join(", ")),
))
}
}
fn normalize_platform_source_visibility(input: &str) -> String {
match input.trim().to_ascii_lowercase().as_str() {
"private" => "private".to_string(),
_ => "public".to_string(),
}
}
fn normalize_hub_grant_scope(input: &str) -> Result<String, PlatformError> {
match input.trim() {
"all_projects" => Ok("all_projects".to_string()),
"selected_project" => Ok("selected_project".to_string()),
_ => Err(PlatformError::new(
"HUB_ACCESS_GRANT_INVALID",
"grant_scope must be all_projects or selected_project",
)),
}
}
fn normalize_limit(value: i64, default_value: i64) -> i64 {
if value <= 0 { default_value } else { value }
}
fn validate_hub_version(version: &str) -> Result<(), PlatformError> {
let valid = !version.is_empty()
&& version.len() <= 80
&& version
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-' | b'_'))
&& !version.contains("..");
if valid {
Ok(())
} else {
Err(PlatformError::new(
"HUB_PUBLISH_INVALID",
"version must be a safe path segment",
))
}
}
fn random_hex(bytes: usize) -> String {
let mut data = vec![0u8; bytes];
rand::rng().fill(data.as_mut_slice());
data.into_iter()
.map(|b| format!("{b:02x}"))
.collect::<String>()
}
fn sha256_hex(input: &[u8]) -> String {
let mut hasher = Sha256::new();
hasher.update(input);
let bytes = hasher.finalize();
bytes.iter().map(|b| format!("{b:02x}")).collect()
}
fn normalize_repo_rel(input: &str) -> String {
let trimmed = input
.trim()
.trim_start_matches("./")
.trim_start_matches('/');
trimmed.replace('\\', "/")
}
fn normalize_template_repo_rel(input: &str) -> String {
let rel = normalize_repo_rel(input);
if rel.is_empty() {
rel
} else if rel.starts_with("pipelines/") {
rel
} else {
format!("pipelines/{rel}")
}
}
fn read_repo_entry(
layout: &ProjectFileLayout,
rel_path: &str,
reason: String,
) -> Result<HubExportEntry, PlatformError> {
let rel_path = normalize_repo_rel(rel_path);
let abs = layout.repo_dir.join(&rel_path);
if !abs.starts_with(&layout.repo_dir) || !abs.is_file() {
return Err(PlatformError::new(
"HUB_ENTRY_MISSING",
format!("file '{}' not found", rel_path),
));
}
let bytes = fs::read(&abs)?;
let size_bytes = bytes.len();
let (encoding, content) = match String::from_utf8(bytes.clone()) {
Ok(text) => ("text".to_string(), text),
Err(_) => (
"base64".to_string(),
base64::engine::general_purpose::STANDARD.encode(bytes),
),
};
Ok(HubExportEntry {
rel_path,
kind: file_kind_from_path(&abs),
size_bytes,
reason,
encoding,
content,
})
}
fn list_repo_folders(layout: &ProjectFileLayout) -> Result<Vec<String>, PlatformError> {
let mut out = Vec::new();
walk_dirs(&layout.repo_dir, &layout.repo_dir, &mut out)?;
out.sort();
Ok(out)
}
fn walk_dirs(root: &Path, current: &Path, out: &mut Vec<String>) -> Result<(), PlatformError> {
let mut entries = fs::read_dir(current)?.collect::<Result<Vec<_>, _>>()?;
entries.sort_by_key(|item| item.path());
for entry in entries {
let path = entry.path();
let name = entry.file_name();
let name = name.to_string_lossy();
if name.starts_with(".git") {
continue;
}
if path.is_dir() {
if let Ok(rel) = path.strip_prefix(root) {
let rel = rel.to_string_lossy().replace('\\', "/");
if !rel.is_empty() {
out.push(rel);
}
}
walk_dirs(root, &path, out)?;
}
}
Ok(())
}
fn collect_tree_entries(
layout: &ProjectFileLayout,
rel_root: &str,
) -> Result<Vec<HubExportEntry>, PlatformError> {
let rel_root = normalize_repo_rel(rel_root);
let base = if rel_root.is_empty() || rel_root == "." {
layout.repo_dir.clone()
} else {
layout.repo_dir.join(&rel_root)
};
if !base.starts_with(&layout.repo_dir) || !base.exists() {
return Err(PlatformError::new(
"HUB_SOURCE_INVALID",
format!("path '{}' not found", rel_root),
));
}
let mut files = Vec::new();
walk_files(&layout.repo_dir, &base, &mut files)?;
Ok(files
.into_iter()
.map(|item| read_repo_entry(layout, &item, "included file".to_string()))
.collect::<Result<Vec<_>, _>>()?)
}
fn walk_files(root: &Path, current: &Path, out: &mut Vec<String>) -> Result<(), PlatformError> {
if current.is_file() {
let rel = current
.strip_prefix(root)
.map_err(|_| PlatformError::new("HUB_SOURCE_INVALID", "path escaped repo root"))?;
out.push(rel.to_string_lossy().replace('\\', "/"));
return Ok(());
}
let mut entries = fs::read_dir(current)?.collect::<Result<Vec<_>, _>>()?;
entries.sort_by_key(|item| item.path());
for entry in entries {
let path = entry.path();
let name = entry.file_name();
let name = name.to_string_lossy();
if name == ".git" {
continue;
}
if path.is_dir() {
walk_files(root, &path, out)?;
} else if path.is_file() {
let rel = path
.strip_prefix(root)
.map_err(|_| PlatformError::new("HUB_SOURCE_INVALID", "path escaped repo root"))?;
out.push(rel.to_string_lossy().replace('\\', "/"));
}
}
Ok(())
}
fn resolve_local_import(
layout: &ProjectFileLayout,
from_rel: &str,
import_src: &str,
) -> Option<String> {
let candidates = if let Some(rest) = import_src.strip_prefix("@/") {
candidate_paths(rest)
} else if import_src.starts_with('.') {
let parent = Path::new(from_rel)
.parent()
.unwrap_or_else(|| Path::new(""));
let joined = parent.join(import_src);
let rel = joined.to_string_lossy().replace('\\', "/");
candidate_paths(&rel)
} else {
return None;
};
for rel in candidates {
let rel = normalize_repo_rel(&rel);
let abs = layout
.repo_pipelines_dir
.join(rel.strip_prefix("pipelines/").unwrap_or(rel.as_str()));
if abs.is_file() {
return Some(normalize_template_repo_rel(&rel));
}
let repo_abs = layout.repo_dir.join(&rel);
if repo_abs.is_file() {
return Some(rel);
}
}
None
}
fn candidate_paths(base: &str) -> Vec<String> {
let base = normalize_repo_rel(base);
let mut out = vec![base.clone()];
if Path::new(&base).extension().is_none() {
for ext in [".tsx", ".ts", ".jsx", ".js", ".css", ".sql", ".json"] {
out.push(format!("{base}{ext}"));
}
for ext in [
"index.tsx",
"index.ts",
"index.jsx",
"index.js",
"index.css",
] {
out.push(format!("{base}/{ext}"));
}
}
out
}
fn file_kind_from_path(path: &Path) -> String {
match path
.extension()
.and_then(|ext| ext.to_str())
.unwrap_or_default()
{
"tsx" => "tsx".to_string(),
"ts" => "ts".to_string(),
"js" => "js".to_string(),
"jsx" => "jsx".to_string(),
"css" => "css".to_string(),
"json" => "json".to_string(),
"sql" => "sql".to_string(),
other if !other.is_empty() => other.to_string(),
_ => "file".to_string(),
}
}
fn install_root_for_target_folder(
package_id: &str,
asset_kind: &str,
target_folder: &str,
) -> String {
let folder = target_folder.trim();
if !folder.is_empty() {
return normalize_install_target_folder(folder, asset_kind);
}
default_install_target_folder(package_id, asset_kind)
}
fn normalize_install_target_folder(target_folder: &str, asset_kind: &str) -> String {
let folder = normalize_repo_rel(target_folder);
if folder.is_empty() {
return ".".to_string();
}
if target_folder.trim_start().starts_with('/')
&& matches!(
asset_kind,
HUB_ASSET_KIND_PIPELINE_BUNDLE | HUB_ASSET_KIND_TEMPLATE_BUNDLE
)
&& !folder.starts_with("pipelines/")
{
return format!("pipelines/{folder}");
}
folder
}
fn install_rel_path_under_folder(install_root: &str, rel_path: &str) -> String {
let root = normalize_repo_rel(install_root);
let rel = install_entry_rel_inside_folder(rel_path);
if root.is_empty() || root == "." {
rel
} else {
format!("{root}/{rel}")
}
}
fn install_entry_rel_inside_folder(rel_path: &str) -> String {
let rel = normalize_repo_rel(rel_path);
if let Some(rest) = rel.strip_prefix("pipelines/") {
rest.to_string()
} else if let Some(rest) = rel.strip_prefix("templates/") {
rest.to_string()
} else {
rel
}
}
fn default_install_target_folder(package_id: &str, asset_kind: &str) -> String {
if matches!(
asset_kind,
HUB_ASSET_KIND_PIPELINE_BUNDLE | HUB_ASSET_KIND_TEMPLATE_BUNDLE
) {
format!("pipelines/hub/{package_id}")
} else if asset_kind == HUB_ASSET_KIND_NODE_BUNDLE {
format!("nodes/{package_id}")
} else {
format!("hub/{package_id}")
}
}
fn write_entry_content(dest_abs: &Path, entry: &HubExportEntry) -> Result<(), PlatformError> {
let bytes = if entry.encoding == "base64" {
base64::engine::general_purpose::STANDARD
.decode(&entry.content)
.map_err(|err| PlatformError::new("HUB_INSTALL", err.to_string()))?
} else {
entry.content.as_bytes().to_vec()
};
fs::write(dest_abs, bytes)?;
Ok(())
}
fn reindex_project_bundle_pipelines(
hub: &HubService,
owner: &str,
project: &str,
entries: &[HubExportEntry],
) -> Result<(), PlatformError> {
for entry in entries {
let rel_path = normalize_repo_rel(&entry.rel_path);
if !rel_path.starts_with("pipelines/") || !rel_path.ends_with(".zf.json") {
continue;
}
let Some(source) = entry_text(entry) else {
continue;
};
let description = serde_json::from_str::<crate::pipeline::PipelineGraph>(&source)
.ok()
.and_then(|graph| graph.description)
.unwrap_or_default();
let trigger_kind = derive_trigger_kind_from_source(&source).unwrap_or_default();
hub.projects.upsert_pipeline_definition(
owner,
project,
&rel_path,
"",
&description,
&trigger_kind,
&source,
)?;
}
Ok(())
}
fn entry_text(entry: &HubExportEntry) -> Option<String> {
if entry.encoding == "text" || entry.encoding.trim().is_empty() {
return Some(entry.content.clone());
}
if entry.encoding == "base64" {
let bytes = base64::engine::general_purpose::STANDARD
.decode(&entry.content)
.ok()?;
return String::from_utf8(bytes).ok();
}
None
}
fn package_policy_entry(rel_path: &str, entry: &HubExportEntry) -> PackagePolicyEntry {
PackagePolicyEntry {
rel_path: rel_path.to_string(),
kind: entry.kind.clone(),
size_bytes: entry.size_bytes,
content: entry_text(entry).unwrap_or_default(),
}
}
fn install_risk_level(policy: &PackageSafetyReview, has_overwrites: bool) -> String {
let mut risk_score = 0;
if has_overwrites {
risk_score += 2;
}
if !policy.credentials_required.is_empty() {
risk_score += 1;
}
if !policy.external_urls.is_empty() {
risk_score += 1;
}
if !policy.database_effects.is_empty() {
risk_score += 2;
}
if !policy.filesystem_effects.is_empty() {
risk_score += 1;
}
if !policy.public_endpoints.is_empty() || !policy.schedules.is_empty() {
risk_score += 2;
}
if !policy.large_files.is_empty() || !policy.seed_data.is_empty() {
risk_score += 1;
}
if risk_score >= 4 {
"high".to_string()
} else if risk_score >= 2 {
"medium".to_string()
} else {
"low".to_string()
}
}
const INITIAL_DATA_DIRS: &[(&str, &str)] = &[
("initial-data/sekejap", "sekejap"),
("initial-data/sqlite", "sqlite"),
("init/sekejap", "sekejap"),
("init/sqlite", "sqlite"),
("seeds/sekejap", "sekejap"),
("seeds/sqlite", "sqlite"),
];
fn initial_data_engine_for_path(rel_path: &str) -> Option<&'static str> {
let rel = normalize_repo_rel(rel_path);
if !rel.ends_with(".sql") {
return None;
}
INITIAL_DATA_DIRS
.iter()
.find_map(|(prefix, engine)| rel.starts_with(&format!("{prefix}/")).then_some(*engine))
}
fn split_initial_data_sql(sql: &str) -> Vec<String> {
let uncommented = sql
.lines()
.filter(|line| !line.trim_start().starts_with("--"))
.collect::<Vec<_>>()
.join("\n");
uncommented
.split(';')
.map(str::trim)
.filter(|stmt| !stmt.is_empty())
.map(|stmt| format!("{stmt};"))
.collect()
}
fn initial_data_step_from_entry(entry: &HubExportEntry) -> Option<HubInitialDataStep> {
let rel = normalize_repo_rel(&entry.rel_path);
let engine = initial_data_engine_for_path(&rel)?;
let sql = entry_text(entry)?;
Some(HubInitialDataStep {
engine: engine.to_string(),
path: rel,
statement_count: split_initial_data_sql(&sql).len(),
size_bytes: entry.size_bytes,
})
}
fn execute_project_initial_data(
data_root: &Path,
owner: &str,
project: &str,
layout: &ProjectFileLayout,
steps: &[HubInitialDataStep],
) -> Result<(), PlatformError> {
for step in steps {
let rel = normalize_repo_rel(&step.path);
let Some(engine) = initial_data_engine_for_path(&rel) else {
return Err(PlatformError::new(
"HUB_INITIAL_DATA_INVALID",
format!("initial data path '{}' is not allowed", step.path),
));
};
if engine != step.engine {
return Err(PlatformError::new(
"HUB_INITIAL_DATA_INVALID",
format!(
"initial data path '{}' does not match engine '{}'",
step.path, step.engine
),
));
}
let path = sanitize_install_repo_path(layout, &rel)?;
let sql = fs::read_to_string(&path)?;
match engine {
"sekejap" => {
for stmt in split_initial_data_sql(&sql) {
sekejap::execute_sql(data_root, owner, project, &stmt, &[], 0, false)?;
}
}
"sqlite" => {
sqlite_schema::execute_sql(data_root, owner, project, &sql)?;
}
_ => {
return Err(PlatformError::new(
"HUB_INITIAL_DATA_INVALID",
format!("unsupported initial data engine '{}'", engine),
));
}
}
}
Ok(())
}
fn collect_initial_data_steps(
repo_dir: &Path,
dir: &Path,
engine: &str,
steps: &mut Vec<HubInitialDataStep>,
) -> Result<(), PlatformError> {
for entry in fs::read_dir(dir)? {
let entry = entry?;
let path = entry.path();
if path.is_dir() {
collect_initial_data_steps(repo_dir, &path, engine, steps)?;
continue;
}
if path
.extension()
.and_then(|value| value.to_str())
.map(|value| value.eq_ignore_ascii_case("sql"))
!= Some(true)
{
continue;
}
let rel = path
.strip_prefix(repo_dir)
.map_err(|err| PlatformError::new("HUB_INITIAL_DATA", err.to_string()))?;
let rel = normalize_repo_rel(&rel.to_string_lossy());
if initial_data_engine_for_path(&rel) != Some(engine) {
continue;
}
let sql = fs::read_to_string(&path)?;
let size_bytes = sql.len();
steps.push(HubInitialDataStep {
engine: engine.to_string(),
path: rel,
statement_count: split_initial_data_sql(&sql).len(),
size_bytes,
});
}
Ok(())
}
fn verify_remote_artifact_hash(payload: &RemoteHubArtifactResponse) -> Result<(), PlatformError> {
if payload.artifact_size_bytes > MAX_REMOTE_HUB_ARTIFACT_BYTES {
return Err(PlatformError::new(
"HUB_ARTIFACT_TOO_LARGE",
"remote artifact exceeds maximum install size",
));
}
let expected = payload
.artifact_sha256
.trim()
.to_string()
.if_empty_then(|| payload.version.artifact_sha256.trim().to_string());
if expected.is_empty() {
return Err(PlatformError::new(
"HUB_REMOTE_HASH_MISSING",
"remote artifact response is missing artifact hash",
));
}
let artifact: HubArtifact = serde_json::from_value(payload.artifact.clone())
.map_err(|err| PlatformError::new("HUB_REMOTE_INVALID", err.to_string()))?;
let bytes = serde_json::to_vec_pretty(&artifact)
.map_err(|err| PlatformError::new("HUB_REMOTE_INVALID", err.to_string()))?;
let actual = sha256_hex(&bytes);
if actual != expected {
return Err(PlatformError::new(
"HUB_REMOTE_HASH_MISMATCH",
"remote artifact hash mismatch",
));
}
Ok(())
}
trait EmptyStringExt {
fn if_empty_then<F>(self, fallback: F) -> String
where
F: FnOnce() -> String;
}
impl EmptyStringExt for String {
fn if_empty_then<F>(self, fallback: F) -> String
where
F: FnOnce() -> String,
{
if self.is_empty() { fallback() } else { self }
}
}
fn sanitize_hub_export_entries(entries: &mut [HubExportEntry]) -> Result<(), PlatformError> {
for entry in entries.iter_mut() {
if normalize_repo_rel(&entry.rel_path) != "zebflow.json" || entry.encoding == "base64" {
continue;
}
let Ok(mut cfg) = serde_json::from_str::<ZebflowJson>(&entry.content) else {
continue;
};
cfg.distribution.hub.producer_enabled = false;
let content = serde_json::to_string_pretty(&cfg)
.map_err(|err| PlatformError::new("HUB_PUBLISH", err.to_string()))?;
entry.size_bytes = content.len();
entry.content = content;
}
Ok(())
}
fn infer_pipeline_meta(source: &str, install_rel: &str) -> (String, String) {
let fallback_title = Path::new(install_rel)
.file_stem()
.and_then(|v| v.to_str())
.unwrap_or("Imported Pipeline")
.replace(".zf", "")
.replace('-', " ");
let Ok(value) = serde_json::from_str::<Value>(source) else {
return (fallback_title, "webhook".to_string());
};
let trigger_kind = value
.get("nodes")
.and_then(Value::as_array)
.and_then(|nodes| {
nodes.iter().find_map(|node| {
node.get("kind")
.and_then(Value::as_str)
.filter(|kind| kind.starts_with("n.trigger."))
.map(|kind| kind.trim_start_matches("n.trigger.").to_string())
})
})
.unwrap_or_else(|| "webhook".to_string());
(fallback_title, trigger_kind)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn normalize_scopes_only_keeps_known_hub_scopes() {
let scopes = normalize_scopes(&[
"hub:publish".to_string(),
" hub:read ".to_string(),
"hub:publish".to_string(),
"custom:other".to_string(),
]);
assert_eq!(
scopes,
vec!["hub:publish".to_string(), "hub:read".to_string()]
);
}
#[test]
fn apply_scope_flags_sets_explicit_permissions() {
let token = apply_scope_flags(HubToken {
token_id: "mkt_1".to_string(),
authority_id: "mka_1".to_string(),
publisher_pk: "mpub_1".to_string(),
owner: "superadmin".to_string(),
project: "default".to_string(),
publisher_id: "zebflow-official".to_string(),
publisher_display_name: "Zebflow Official".to_string(),
publisher_url: "/publishers/zebflow-official".to_string(),
publisher_email: "publishers@zebflow.com".to_string(),
title: "Official".to_string(),
secret_hash: "hash".to_string(),
scopes: vec!["hub:read".to_string(), "hub:publish".to_string()],
scope_read: false,
scope_publish: false,
scope_manage: false,
expires_at: None,
last_used_at: None,
revoked_at: None,
created_at: 0,
updated_at: 0,
});
assert!(token.scope_read);
assert!(token.scope_publish);
assert!(!token.scope_manage);
assert!(token.grants_scope("hub:read"));
assert!(token.grants_scope("hub:publish"));
assert!(!token.grants_scope("hub:manage"));
}
#[test]
fn publisher_scopes_respect_current_permissions() {
let publisher = HubPublisher {
authority_id: DEFAULT_HUB_SERVICE_INSTANCE_ID.to_string(),
publisher_pk: "mpub_1".to_string(),
owner: HUB_SERVICE_SCOPE_OWNER.to_string(),
project: HUB_SERVICE_SCOPE_PROJECT.to_string(),
publisher_id: "zebflow-official".to_string(),
display_name: "Zebflow Official".to_string(),
publisher_url: "/publishers/zebflow-official".to_string(),
email: "publishers@zebflow.com".to_string(),
description: String::new(),
icon_url: String::new(),
website_url: String::new(),
enabled: true,
can_read: true,
can_publish: false,
can_manage: false,
max_packages: 20,
max_package_bytes: 10 * 1024 * 1024,
max_media_files: 8,
max_image_bytes: 2 * 1024 * 1024,
created_at: 0,
updated_at: 0,
};
assert!(validate_publisher_scopes(&publisher, &["hub:read".to_string()]).is_ok());
let err = validate_publisher_scopes(&publisher, &["hub:publish".to_string()])
.expect_err("publish scope should be denied");
assert_eq!(err.code, "HUB_PUBLISHER_SCOPE_DENIED");
}
#[test]
fn hub_versions_are_single_safe_path_segments() {
assert!(validate_hub_version("1.0.0").is_ok());
assert!(validate_hub_version("1.0.0-beta_1").is_ok());
assert!(validate_hub_version("../secret").is_err());
assert!(validate_hub_version("1/2").is_err());
assert!(validate_hub_version("release candidate").is_err());
}
#[test]
fn ownerless_hub_api_base_builds_hub_routes() {
let repo = PlatformHubRepository {
source_id: "pmr_1".to_string(),
owner_user_id: "user_1".to_string(),
owner: "superadmin".to_string(),
repository_id: "zebflow-hub".to_string(),
title: "Zebflow Hub".to_string(),
base_url: "https://hub.zebflow.com/api".to_string(),
remote_owner: String::new(),
remote_project: String::new(),
read_token: String::new(),
visibility: "public".to_string(),
enabled: true,
created_at: 0,
updated_at: 0,
};
assert_eq!(
remote_hub_url_for_platform(&repo, "remote/assets"),
"https://hub.zebflow.com/api/hub/remote/assets"
);
}
}