/**
 * Interface for managing secrets in Flink.
 * Delegates to external secret management systems.
 */
@PublicEvolving
public interface SecretStore {
}


/**
 * Interface for managing secrets in Flink.
 * Delegates to external secret management systems.
 */
@PublicEvolving
public interface ReadableSecretStore extends SecretStore {
    /**
     * Retrieves secret data by identifier.
     * 
     * @param secretId The unique identifier of the secret
     * @return The secret data
     * @throws SecretNotFoundException if the secret doesn't exist
     */
    Map<String, String> getSecret(String secretId) throws SecretNotFoundException;
}


/**
 * Interface for managing secrets in Flink.
 * Delegates to external secret management systems.
 */
@PublicEvolving
public interface WritableSecretStore extends SecretStore {
    /**
     * Stores secret data and returns a unique identifier.
     * 
     * @param secretData The secret key-value pairs to store
     * @return A unique identifier for retrieving the secret
     */
    String storeSecret(Map<String, String> secretData);
   /**
     * Removes a secret from the store.
     * 
     * @param secretId The unique identifier of the secret
     */
    void removeSecret(String secretId);
	/**
     * Atomically updates an existing secret with new data.
     * 
     * @param secretId The existing secret identifier
     * @param newSecretData The new secret data
     * @throws SecretNotFoundException if the secret doesn't exist
     */
    void updateSecret(String secretId, Map<String, String> newSecretData) 
        throws SecretNotFoundException;
}

/**
 * Factory for creating SecretStore instances.
 */
@PublicEvolving
public interface SecretStoreFactory extends Factory {
    
    /**
     * Creates a SecretStore instance.
     */
    SecretStore createSecretStore();

    /**
     * Initialize secret store.
     */
    void open(Context context) throws Exception;

    /**
     * Close secret store
     */
    void close() throws CatalogException;

    interface Context {
        /**
         * Returns the options with which the secret store is created.
         *
         * <p>An implementation should perform validation of these options.
         */
        Map<String, String> getOptions();

        /** Gives read-only access to the configuration of the current session. */
        ReadableConfig getConfiguration();

        /**
         * Returns the class loader of the current session.
         *
         * <p>The class loader is in particular useful for discovering further (nested) factories.
         */
        ClassLoader getClassLoader();
    }
}

/**
 * Interface for a connection in a catalog.
 * Contains non-sensitive configuration and references to secrets.
 */
@PublicEvolving
public interface CatalogConnection {
    
    /**
     * Returns non-sensitive connection options.
     */
    Map<String, String> getOptions();
    
    /**
     * Get comment of the connection.
     */
    String getComment();
    
    /**
     * Get a deep copy of the CatalogConnection instance.
     */
    CatalogConnection copy();
}

/**
 * Interface for a connection that may contain sensitive data
 */
@PublicEvolving
public interface SensitiveConnection {
    
    /**
     * Returns connection options which contains plain secrets.
     */
    Map<String, String> getOptions();
    
    /**
     * Get comment of the connection.
     */
    String getComment();
}    

public final class ObjectIdentifier implements Serializable {
    public of(String objectName);
    
    boolean hasOnlyObjectName();
}

public interface CatalogTable {
    Optional<ObjectIdentifier> getConnection();
}

/**
 * Factory for processing connection configurations.
 * Extracts secrets during creation, resolves them during usage.
 */
@PublicEvolving
public interface ConnectionFactory extends Factory {
    
    /**
     * Processes raw connection options and returns a catalog connection.
     * Extracts sensitive data to SecretStore during processing.
     * 
     * @param connectionToEncrypt Raw connection  (may contain sensitive data)
     * @param secretStore The SecretStore to use for storing secrets
     * @return CatalogConnection with secrets extracted
     */
    CatalogConnection createConnection(
        SensitiveConnection connectionToEncrypt, 
        WritableSecretStore secretStore);
    
    /**
     * Resolves a connection for usage by merging catalog data with secrets.
     * 
     * @param connectionToDecrypt The catalog connection with secret references
     * @param secretStore The SecretStore to retrieve secrets from
     * @return Complete connection configuration with resolved secrets
     */
    SensitiveConnection resolveConnection(
        CatalogConnection connectionToDecrypt, 
        ReadableSecretStore secretStore);
}

public interface Catalog {
    // ... existing methods ...
    
    /**
     * Get names of all connections under this database.
     */
    default List<String> listConnections(String databaseName) 
            throws DatabaseNotExistException, CatalogException {
        return Collections.emptyList();
    }
    
    /**
     * Returns a CatalogConnection identified by the given ObjectPath.
     * Uses 3-part identifier: catalog.database.connection
     */
    default CatalogConnection getConnection(ObjectPath connectionPath) 
            throws ConnectionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }
    
    /**
     * Creates a new connection.
     */
    default void createConnection(ObjectPath connectionPath, CatalogConnection connection, boolean ignoreIfExists) 
            throws ConnectionAlreadyExistException, DatabaseNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }
    
    /**
     * Modifies an existing connection.
     */
    default void alterConnection(ObjectPath connectionPath, CatalogConnection newConnection, boolean ignoreIfNotExists) 
            throws ConnectionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }
    
    /**
     * Drop a connection.
     */
    default void dropConnection(ObjectPath connectionPath, boolean ignoreIfNotExists) 
            throws ConnectionNotExistException, CatalogException {
        throw new UnsupportedOperationException();
    }
    
    // Additional methods: connectionExists, renameConnection...
}

public class EnvironmentSettings {

    Class Builder {
      /**
       * Configures SecretStore using factory pattern.
       */
      public Builder withSecretStore(SecretStore secretStore) {
          // Builder with SecretStore implementation
      }
    }
}

public class ConnectionDescriptor {
    
    public Map<String, String> getOptions();
    public String getComment();
    public Builder toBuilder();
    public Builder forType(String type);

    public 
    public static class Builder {
        private Map<String, String> connectionOptions;
        private @Nullable String comment;
        
        public Builder();
        public Builder(ConnectionDescriptor);
        
        public void option(String key, String value);
        public void option(ConfigOption<T> configOption, T value);
        public void comment();
        public ConnectionDescriptor build();
    }
}

public class TableEnvironment {
    public void createConnection(String path, ConnectionDescriptor connectionDescriptor);
    public void createConnection(String path, ConnectionDescriptor connectionDescriptor, boolean ignoreIfExists);
    public void createTemporaryConnection(String path, ConnectionDescriptor connectionDescriptor);
    public void createTemporaryConnection(String path, ConnectionDescriptor connectionDescriptor, boolean ignoreIfExists);
    public void createTemporarySystemConnection(String path, ConnectionDescriptor connectionDescriptor);
    public void createTemporarySystemConnection(String path, ConnectionDescriptor connectionDescriptor, boolean ignoreIfExists);
    
    public boolean dropConnection(String path);
    public boolean dropConnection(String path, boolean ignoreIfNotExists);
    public boolean dropTemporaryConnection(String path); 
    public boolean dropTemporarySystemConnection(String path);
}

CREATE [TEMPORARY] [SYSTEM] CONNECTION [IF NOT EXISTS] [catalog_name.][db_name.]connection_name 
[COMMENT connection_comment] 
WITH (
    'type' = 'basic' | 'bearer' | 'oauth' | 'custom_type',
    'url' = 'endpoint_url',
    'username' = 'user_name',     -- Will be extracted to SecretStore
    'password' = 'user_password', -- Will be extracted to SecretStore
    ...
)

-- Using CONNECTION keyword in DDL
CREATE TABLE customers (
    customer_id INT,
    name STRING,
    PRIMARY KEY (customer_id) NOT ENFORCED
) USING CONNECTION mycat.mydb.mysql_connection
WITH (
    'table-name' = 'customers',
    'lookup.cache' = 'PARTIAL'
);

-- Alter connection
ALTER CONNECTION mycat.mydb.mysql_connection SET (
    'password' = 'new_password'  -- Updates secret in SecretStore
);

-- Drop connection  
DROP CONNECTION [IF EXISTS] mycat.mydb.mysql_connection;

-- Describe connection (secrets are redacted)
DESCRIBE CONNECTION mycat.mydb.mysql_connection;

-- Show connections
SHOW CONNECTIONS [LIKE 'pattern'] [FROM catalog_name.db_name];

-- Show create connection
SHOW CREATE CONNECTION <connection_name>

public class DefaultConnectionFactory implements ConnectionFactory {
    
    public static final String IDENTIFIER = "default";
    // This conforms to what's defined in GlobalConfiguration      
    public static final Set<String> SECRET_FIELDS = Set.of("password", "secret", "fs.azure.account.key", "apikey", "api-key", "auth-params", "service-key", "token", "basic-auth", "jaas.config", "http-headers");
    private static final String SECRET_KEY = "encrypted_secret_key";
    
    @Override
    public String factoryIdentifier() {
        return IDENTIFIER;
    }
    
    @Override
    public CatalogConnection createConnection(Map<String, String> options, SecretStore secretStore) {
        // Extract secrets
        Map<String, String> secrets = new HashMap<>();
        Map<String, String> publicOptions = new HashMap<>(options);
        
        for (String secretField : SECRET_FIELDS) {
            if (options.containsKey(secretField)) {
                secrets.put(secretField, options.get(secretField));
                publicOptions.remove(secretField);
            }
        }
        
        // Store secrets if any
        String secretId = secrets.isEmpty() ? null : secretStore.storeSecret(secrets);
        if (secretId != null) {
            publicOptions.put(SECRET_KEY, secretId);
        }
        
        return new DefaultCatalogConnection(publicOptions, null);
    }
    
    @Override
    public SensitiveConnection resolveConnection(CatalogConnection connection, SecretStore secretStore) {
        Map<String, String> resolved = new HashMap<>(connection.getOptions());
        
        // Merge secrets if they exist
        if (connection.getOptions().containsKey(SECRET_KEY)) {
            Map<String, String> secrets = secretStore.getSecret(connection.getOptions().get(SECRET_KEY));
            resolved.putAll(secrets);
            resolved.remove(SECRET_KEY);
        }
        
        return new DefaultSensitiveConnection(resolved, null);
    }
}

-- Create connection with username/password (secrets extracted automatically)
CREATE CONNECTION mycat.mydb.mysql_prod WITH (
    'type' = 'default',
    'hostname' = 'jdbc:mysql://prod-db:3306/orders',
    'port' = '1234',
    'username' = 'prod_user',    -- Moved to SecretStore
    'password' = 'secret123'     -- Moved to SecretStore
);

-- Catalog stores:
-- {
--   'type' = 'default',
--   'hostname' = 'jdbc:mysql://prod-db:3306/orders',
--   'username' = 'prod_user',
--   'secret.id' = 'conn-12345'
-- }
-- SecretStore stores: {password' = 'secret123'}

-- Use connection
CREATE TABLE orders (
    order_id INT,
    customer_id INT,
    amount DECIMAL(10,2)
) USING CONNECTION mycat.mydb.mysql_prod
WITH (
    'connector' = 'jdbc',
    'tables' = 'orders'
    ...
);

public class AzureKeyVaultSecretStoreFactory implements SecretStoreFactory {
    
    public static final String IDENTIFIER = "azure-key-vault";
    
    @Override
    public String factoryIdentifier() {
        return IDENTIFIER;
    }
    
    @Override
    public SecretStore createSecretStore(Context context) {
        ReadableConfig options = context.getOptions();
        String vaultUrl = options.get(VAULT_URL);
        String clientId = options.get(CLIENT_ID);
        
        return new AzureKeyVaultSecretStore(vaultUrl, clientId);
    }
}

# META-INF/services/org.apache.flink.table.factories.SecretStoreFactory
com.azure.flink.AzureKeyVaultSecretStoreFactory
com.aws.flink.AwsSecretsManagerSecretStoreFactory
org.apache.flink.table.store.InMemorySecretStoreFactory

// Configure SecretStore in Table API
EnvironmentSettings settings = EnvironmentSettings
    .withSecretStore(new AzureSecretStore());

TableEnvironment tableEnv = TableEnvironment.create(settings);

// Configure SecretStore in yaml file
table.secret-store.kind: azure-key-vault
table.secret-store.azure-key-vault.vault.url: https://company-vault.vault.azure.net/

// Create connection
tableEnv.createConnection("mycat.mydb.mysql_prod", 
    ConnectionDescriptor.forType("basic")
        .option("url", "jdbc:mysql://prod:3306/db")
        .option("username", "user")    // Will be moved to SecretStore
        .option("password", "pass")    // Will be moved to SecretStore
        .build());

// Use connection in table
tableEnv.createTable("orders",
    TableDescriptor.forConnector("jdbc")
        .connection("mycat.mydb.mysql_prod")  // Reference to connection
        .option("table-name", "orders")
        .schema(Schema.newBuilder()...build())
        .build());