backends now fully support branching

Author: AlexMercedCoder

docs/authentication.md

@@ -79,16 +79,36 @@ For programmatic access (CI/CD, ETL, automation), use **Service Users**.
 
 ## 🐍 Connecting with PyIceberg
 
-### Option A: Manual Token (Access Token)
+### Option A: Standard OAuth2 Credential Flow (Recommended)
+Use a Service User's `client_id` (UUID) and `client_secret` (API Key) for automated authentication.
+
+```python
+from pyiceberg.catalog import load_catalog
+
+catalog = load_catalog(
+    "pangolin",
+    **{
+        "uri": "http://localhost:8080/v1/default/",
+        "type": "rest",
+        "credential": "<service_user_uuid>:<api_key>",
+        "oauth2-server-uri": "http://localhost:8080/v1/default/v1/oauth/tokens",
+        "scope": "catalog"
+    }
+)
+```
+
+> [!NOTE]
+> The `oauth2-server-uri` must be the full path. Pangolin supports paths starting with both `/v1/...` and `/api/v1/iceberg/...`.
+
+### Option B: Manual Token (Access Token)
 If you have a JWT from a previous login:
 ```python
 from pyiceberg.catalog import load_catalog
 
 catalog = load_catalog(
     "pangolin",
     **{
-        "uri": "http://localhost:8080",
-        "prefix": "analytics",
+        "uri": "http://localhost:8080/v1/default/",
         "token": "YOUR_JWT_TOKEN",
     }
 )

docs/features/pyiceberg_testing.md

@@ -153,14 +153,20 @@ from pyiceberg.catalog import load_catalog
 catalog = load_catalog(
     "pangolin",
     **{
-        "uri": "http://localhost:8080/api/v1/iceberg/default", 
+        "uri": "http://localhost:8080/v1/default/", 
         "type": "rest",
-        "credential": "<service_user_id>:<service_user_api_key>",
-        "oauth2-server-uri": "http://localhost:8080/v1/rest/v1/oauth/tokens",
+        "credential": "service_user_uuid:service_user_api_key",
+        "oauth2-server-uri": "http://localhost:8080/v1/default/v1/oauth/tokens",
+        "scope": "catalog"
     }
 )
 ```
 
+> [!NOTE]
+> *   **URI Suffix**: Ensure your `uri` ends with a trailing slash `/`.
+> *   **Auth Endpoint**: `oauth2-server-uri` must be the full path: `{base_url}/v1/{catalog_name}/v1/oauth/tokens`. Pangolin also supports the `/api/v1/iceberg/{catalog_name}/v1/oauth/tokens` prefix for compatibility.
+> *   **Credential Format**: Must be `client_id:client_secret` (UUID:API_Key).
+
 ### Option 2: Using an Existing Token (Temporary)
 
 If you have a JWT token from the UI or another login flow, you can use it directly.

docs/features/service_users.md

@@ -13,10 +13,12 @@ Administrators can provision and rotate keys directly from the terminal.
 ```bash
 pangolin-admin create-service-user \
   --name "etl-pipeline" \
-  --role "TenantUser" \
-  --expires-in-days 365
+  --role "TenantUser"
 ```
 
+> [!NOTE]
+> Ensure you record the **Service User ID (UUID)** from the output. If the CLI does not display the ID, use the Management UI or the API directly to retrieve it after creation.
+
 **Rotate API Key:**
 ```bash
 pangolin-admin rotate-service-user-key --id <service-user-uuid>
@@ -47,9 +49,10 @@ from pyiceberg.catalog import load_catalog
 catalog = load_catalog(
     "pangolin",
     **{
-        "uri": "https://api.pangolin.io/api/v1/iceberg/default",
+        "uri": "http://localhost:8080/v1/default/",
         "credential": "<service_user_uuid>:<api_key>",
-        "oauth2-server-uri": "https://api.pangolin.io/v1/rest/v1/oauth/tokens",
+        "oauth2-server-uri": "http://localhost:8080/v1/default/v1/oauth/tokens",
+        "scope": "catalog",
         "type": "rest",
     }
 )

pangolin/pangolin_api/src/iceberg/oauth.rs

@@ -19,19 +19,29 @@ use crate::auth::Claims;
 use crate::error::ApiError;
 use crate::iceberg::AppState;
 
-#[derive(Deserialize)]
+use utoipa::ToSchema;
+
+#[derive(Deserialize, ToSchema)]
 pub struct OAuthTokenRequest {
+    #[schema(example = "client_credentials")]
     grant_type: String,
+    #[schema(example = "a1b2c3d4-e5f6-7890-1234-567890abcdef")]
     client_id: String,
+    #[schema(example = "pgl_...")]
     client_secret: String,
+    #[schema(example = "catalog")]
     scope: Option<String>,
 }
 
-#[derive(Serialize)]
+#[derive(Serialize, ToSchema)]
 pub struct OAuthTokenResponse {
+    #[schema(example = "eyJhbGciOiJIUzI1Ni...")]
     access_token: String,
+    #[schema(example = "Bearer")]
     token_type: String,
+    #[schema(example = 3600)]
     expires_in: u64,
+    #[schema(example = "urn:ietf:params:oauth:token-type:access_token")]
     issued_token_type: String,
 }
 
@@ -45,6 +55,18 @@ pub struct OAuthTokenResponse {
 /// - `client_secret` -> Service User API Key
 /// 
 /// If valid, it returns a standard Pangolin JWT signed by the server key.
+#[utoipa::path(
+    post,
+    path = "/v1/{prefix}/v1/oauth/tokens",
+    operation_id = "oauth_token",
+    request_body(content = OAuthTokenRequest, content_type = "application/x-www-form-urlencoded"),
+    responses(
+        (status = 200, description = "Token issued successfully", body = OAuthTokenResponse),
+        (status = 400, description = "Invalid grant_type or format"),
+        (status = 401, description = "Invalid client credentials")
+    ),
+    tag = "Authentication"
+)]
 pub async fn handle_oauth_token(
     State(store): State<AppState>,
     Form(payload): Form<OAuthTokenRequest>,

pangolin/pangolin_api/src/iceberg/tables.rs

@@ -336,13 +336,15 @@ pub async fn create_table(
                  return (StatusCode::INTERNAL_SERVER_ERROR, "Failed to write metadata").into_response();
             }
 
-            let _ = store.log_audit_event(tenant_id, pangolin_core::audit::AuditLogEntry::legacy_new(
+            let _ = store.log_audit_event(tenant_id, pangolin_core::audit::AuditLogEntry::success(
                 tenant_id,
+                Some(session.user_id),
                 session.username.clone(),
-                "create_table".to_string(),
+                pangolin_core::audit::AuditAction::CreateTable,
+                pangolin_core::audit::ResourceType::Table,
+                Some(asset.id),
                 format!("{}/{}/{}", catalog_name, ns_name, tbl_name),
-                Some(location.clone())
-            )).await;
+            ).with_metadata(serde_json::json!({ "location": location.clone() }))).await;
 
             let credentials = match store.get_catalog(tenant_id, catalog_name.clone()).await {
                 Ok(Some(c)) => {
@@ -667,13 +669,15 @@ pub async fn update_table(
 
         match store.update_metadata_location(tenant_id, &catalog_name, Some(branch.clone()), namespace_parts.clone(), table_name.clone(), current_metadata_location.clone(), new_metadata_location.clone()).await {
             Ok(_) => {
-                let _ = store.log_audit_event(tenant_id, pangolin_core::audit::AuditLogEntry::legacy_new(
+                let _ = store.log_audit_event(tenant_id, pangolin_core::audit::AuditLogEntry::success(
                     tenant_id,
+                    Some(session.user_id),
                     session.username.clone(),
-                    "update_table".to_string(),
+                    pangolin_core::audit::AuditAction::UpdateTable,
+                    pangolin_core::audit::ResourceType::Table,
+                    Some(asset.id),
                     format!("{}/{}/{}", catalog_name, namespace, table_name),
-                    Some(new_metadata_location.clone())
-                )).await;
+                ).with_metadata(serde_json::json!({ "new_metadata_location": new_metadata_location.clone() }))).await;
 
                 return (StatusCode::OK, Json(TableResponse::new(
                     Some(new_metadata_location.clone()),
@@ -740,12 +744,14 @@ pub async fn rename_table(
 
     match store.rename_asset(tenant_id, &catalog_name, branch, source_ns.clone(), source_name.clone(), dest_ns.clone(), dest_name.clone()).await {
         Ok(_) => {
-            let _ = store.log_audit_event(tenant_id, pangolin_core::audit::AuditLogEntry::legacy_new(
+            let _ = store.log_audit_event(tenant_id, pangolin_core::audit::AuditLogEntry::success(
                 tenant_id,
+                Some(session.user_id),
                 session.username.clone(),
-                "rename_table".to_string(),
+                pangolin_core::audit::AuditAction::RenameTable,
+                pangolin_core::audit::ResourceType::Table,
+                None, // Cannot determine asset ID easily without lookup
                 format!("{}/{}.{} -> {}.{}", catalog_name, source_ns.join("."), source_name, dest_ns.join("."), dest_name),
-                None
             )).await;
 
             StatusCode::NO_CONTENT.into_response()

pangolin/pangolin_api/src/merge_handlers.rs

@@ -253,8 +253,8 @@ pub async fn complete_merge(
     match store.merge_branch(
         operation.tenant_id,
         &operation.catalog_name,
-        operation.source_branch.clone(),
         operation.target_branch.clone(),
+        operation.source_branch.clone(),
     ).await {
         Ok(_) => {
             // TODO: merge_branch should return the commit ID

pangolin/pangolin_api/src/openapi.rs

@@ -180,6 +180,9 @@ use pangolin_core::business_metadata::{BusinessMetadata, AccessRequest, RequestS
         iceberg::tables::perform_maintenance,
         iceberg::tables::table_exists,
         iceberg::namespaces::list_namespaces_tree,
+        
+        // Iceberg OAuth
+        iceberg::oauth::handle_oauth_token,
 
         // Signing / Credential Vending
         signing_handlers::get_table_credentials,
@@ -219,6 +222,9 @@ use pangolin_core::business_metadata::{BusinessMetadata, AccessRequest, RequestS
             service_user_handlers::CreateServiceUserRequest, service_user_handlers::UpdateServiceUserRequest,
             oauth_handlers::OAuthCallback, oauth_handlers::AuthorizeParams,
             
+            // Iceberg OAuth types
+            iceberg::oauth::OAuthTokenRequest, iceberg::oauth::OAuthTokenResponse,
+            
             // Branch/Tag/Merge types
             pangolin_handlers::CreateBranchRequest, pangolin_handlers::ListBranchParams, pangolin_handlers::MergeBranchRequest, pangolin_handlers::BranchResponse,
             merge_handlers::ResolveConflictRequest, merge_handlers::MergeOperationResponse,

pangolin/pangolin_api/src/pangolin_handlers.rs

@@ -137,6 +137,7 @@ pub async fn create_branch(
 
     let scope = PermissionScope::Catalog { catalog_id: catalog.id };
 
+    // Check permission
     match crate::authz::check_permission(&store, &session, &required_action, &scope).await {
         Ok(true) => (),
         Ok(false) => return (StatusCode::FORBIDDEN, "Forbidden").into_response(),
@@ -150,15 +151,13 @@ pub async fn create_branch(
     // 2. If assets are specified, copy them from `from_branch`.
 
     let mut branch_assets = vec![];
+    tracing::info!("create_branch called for branch: {}, from: {}", payload.name, from_branch);
+    
+    // ...
 
     if let Some(assets_to_copy) = &payload.assets {
+        tracing::info!("Explicit asset list provided: {} assets", assets_to_copy.len());
         for asset_name in assets_to_copy {
-            // We need the namespace for the asset. 
-            // The current API `CreateBranchRequest` only lists asset names, but assets are scoped by namespace.
-            // This is a limitation. The user request said "specified tables".
-            // Assuming for now that `assets` contains "namespace.table" strings or we need to change the request to be more structured.
-            // Let's assume "namespace.table" format for simplicity in this MVP.
-            
             let parts: Vec<&str> = asset_name.split('.').collect();
             if parts.len() < 2 {
                 continue; // Skip invalid format
@@ -168,18 +167,38 @@ pub async fn create_branch(
 
             // Get asset from source branch
             if let Ok(Some(asset)) = store.get_asset(tenant_id, catalog_name, Some(from_branch.to_string()), namespace_parts.clone(), table_name.clone()).await {
-                // Create asset in new branch
-                if let Ok(_) = store.create_asset(tenant_id, catalog_name, Some(payload.name.clone()), namespace_parts, asset).await {
+                // Create asset in new branch with NEW ID to avoid unique constraint violation
+                let mut new_asset = asset.clone();
+                new_asset.id = uuid::Uuid::new_v4();
+                if let Ok(_) = store.create_asset(tenant_id, catalog_name, Some(payload.name.clone()), namespace_parts, new_asset).await {
                     branch_assets.push(asset_name.clone());
                 }
             }
         }
     } else {
-        // If no assets specified, maybe we copy ALL assets? 
-        // Or create an empty branch?
-        // The user said "a branch shouldn't apply to the whole catalog but only to specified tables".
-        // This implies if you don't specify tables, you might get an empty branch or it's an error?
-        // Let's assume empty branch if None.
+        tracing::info!("No explicit assets. Auto-propagating from {}", from_branch);
+        // ...
+        if let Ok(namespaces) = store.list_namespaces(tenant_id, catalog_name, None).await {
+            tracing::info!("Found {} namespaces", namespaces.len());
+            for ns in namespaces {
+                if let Ok(assets) = store.list_assets(tenant_id, catalog_name, Some(from_branch.to_string()), ns.name.clone()).await {
+                    tracing::info!("Found {} assets in namespace {:?} on branch {}", assets.len(), ns.name, from_branch);
+                    for asset in assets {
+                        let mut new_asset = asset.clone();
+                        new_asset.id = uuid::Uuid::new_v4();
+                        match store.create_asset(tenant_id, catalog_name, Some(payload.name.clone()), ns.name.clone(), new_asset.clone()).await {
+                            Ok(_) => {
+                                tracing::info!("Copied asset {} to branch {}", asset.name, payload.name);
+                                branch_assets.push(format!("{}.{}", ns.name.join("."), asset.name));
+                            },
+                            Err(e) => tracing::error!("Failed to copy asset {}: {}", asset.name, e),
+                        }
+                    }
+                } else {
+                    tracing::warn!("Failed to list assets in namespace {:?}", ns.name);
+                }
+            }
+        }
     }
 
     let branch = Branch {
@@ -334,7 +353,7 @@ pub async fn merge_branch(
     }
 
     // No conflicts - proceed with merge
-    match store.merge_branch(tenant_id, catalog_name, payload.source_branch.clone(), payload.target_branch.clone()).await {
+    match store.merge_branch(tenant_id, catalog_name, payload.target_branch.clone(), payload.source_branch.clone()).await {
         Ok(_) => {
             // Complete the merge operation
             let commit_id = uuid::Uuid::new_v4(); // In real implementation, get actual commit ID
@@ -349,11 +368,13 @@ pub async fn merge_branch(
                 None
             )).await;
 
-            (StatusCode::OK, Json(serde_json::json!({
-                "status": "merged",
-                "operation_id": operation.id,
-                "commit_id": commit_id
-            }))).into_response()
+            // Return full updated operation object to satisfy SDK Pydantic model
+            let mut result_op = operation.clone();
+            result_op.status = pangolin_core::model::MergeStatus::Completed;
+            result_op.result_commit_id = Some(commit_id);
+            result_op.completed_at = Some(chrono::Utc::now());
+
+            (StatusCode::OK, Json(result_op)).into_response()
         },
         Err(e) => {
             let _ = store.abort_merge_operation(operation.id).await;
@@ -720,6 +741,17 @@ pub async fn create_catalog(
                 tracing::warn!("Failed to create default namespace for catalog {}: {}", catalog.name, e);
             }
 
+            // Create 'main' branch
+             let main_branch = pangolin_core::model::Branch {
+                name: "main".to_string(),
+                head_commit_id: None,
+                branch_type: pangolin_core::model::BranchType::Ingest,
+                assets: vec![],
+             };
+            if let Err(e) = store.create_branch(tenant_id, &catalog.name, main_branch).await {
+                tracing::warn!("Failed to create main branch for catalog {}: {}", catalog.name, e);
+            }
+
             // Audit Log
             let _ = store.log_audit_event(tenant_id, pangolin_core::audit::AuditLogEntry::legacy_new(
                 tenant_id,

pangolin/pangolin_store/migrations/20251221000000_fix_asset_branching.sql

@@ -9,3 +9,4 @@ ALTER TABLE assets ALTER COLUMN branch_name SET NOT NULL;
 -- Drop and recreate PK to include branch_name
 ALTER TABLE assets DROP CONSTRAINT assets_pkey;
 ALTER TABLE assets ADD PRIMARY KEY (tenant_id, catalog_name, branch_name, namespace_path, name);
+

pangolin/pangolin_store/migrations/sqlite/001_initial.sql

@@ -65,7 +65,7 @@ CREATE INDEX IF NOT EXISTS idx_namespaces_tenant_catalog ON namespaces(tenant_id
 
 -- Assets (Added ID)
 CREATE TABLE IF NOT EXISTS assets (
-    id TEXT PRIMARY KEY,
+    id TEXT,
     tenant_id TEXT NOT NULL,
     catalog_name TEXT NOT NULL,
     namespace_path TEXT NOT NULL, -- JSON array