Rebase with github repository

Connected.ServiceModel.Client.Data/AggregatedCommandBuilder.cs

@@ -0,0 +1,50 @@
+using System.Collections.Immutable;
+using Connected.Entities;
+using Connected.Entities.Storage;
+
+namespace Connected.ServiceModel.Client.Data;
+internal sealed class AggregatedCommandBuilder<TEntity>
+{
+	public StorageOperation? Build(TEntity entity)
+	{
+		if (entity is not IEntity ie)
+			throw new ArgumentException(nameof(entity));
+
+		switch (ie.State)
+		{
+			case State.New:
+				return BuildInsert(ie);
+			case State.Default:
+				return BuildUpdate(ie);
+			case State.Deleted:
+				return BuildDelete(ie);
+			default:
+				throw new NotSupportedException();
+		}
+	}
+
+	public List<StorageOperation> Build(ImmutableArray<TEntity> entities)
+	{
+		var result = new List<StorageOperation>();
+
+		foreach (var entity in entities)
+			result.Add(Build(entity));
+
+		return result;
+	}
+
+	private StorageOperation? BuildInsert(IEntity entity)
+	{
+		return new InsertCommandBuilder().Build(entity);
+	}
+
+	private StorageOperation? BuildUpdate(IEntity entity)
+	{
+		return new UpdateCommandBuilder().Build(entity);
+	}
+
+	private StorageOperation? BuildDelete(IEntity entity)
+	{
+		return new DeleteCommandBuilder().Build(entity);
+	}
+}

Connected.ServiceModel.Client.Data/Boot.cs

@@ -0,0 +1,15 @@
+using Connected.Annotations;
+using Connected.ServiceModel.Client.Data.Remote;
+using Microsoft.Extensions.DependencyInjection;
+
+[assembly: MicroService(MicroServiceType.Provider)]
+
+namespace Connected.ServiceModel.Client.Data;
+
+internal sealed class Boot : Startup
+{
+	protected override void OnConfigureServices(IServiceCollection services)
+	{
+		services.AddScoped<RemoteTableService>();
+	}
+}

Connected.ServiceModel.Client.Data/CommandBuilder.cs

@@ -0,0 +1,206 @@
+using System.Data;
+using System.Reflection;
+using System.Text;
+using Connected.Entities;
+using Connected.Entities.Annotations;
+using Connected.Entities.Storage;
+using Connected.Interop;
+
+namespace Connected.ServiceModel.Client.Data;
+
+internal abstract class CommandBuilder
+{
+	private readonly List<IStorageParameter> _parameters;
+	private readonly List<PropertyInfo> _whereProperties;
+	private List<PropertyInfo> _properties;
+
+	protected CommandBuilder()
+	{
+		_parameters = new List<IStorageParameter>();
+		_whereProperties = new List<PropertyInfo>();
+
+		Text = new StringBuilder();
+	}
+
+	public StorageOperation? Build(IEntity entity)
+	{
+		Entity = entity;
+
+		if (TryGetExisting(out StorageOperation? existing))
+		{
+			/*
+			 * We need to rebuild an instance since StorageOperation
+			 * is immutable
+			 */
+			var result = new StorageOperation
+			{
+				CommandText = existing.CommandText,
+				CommandTimeout = existing.CommandTimeout,
+				CommandType = existing.CommandType,
+				Concurrency = existing.Concurrency
+			};
+
+			if (result.Parameters is null)
+				return result;
+
+			foreach (var parameter in result.Parameters)
+			{
+				if (parameter.Direction == ParameterDirection.Input)
+				{
+					if (ResolveProperty(parameter.Name) is PropertyInfo property)
+					{
+						result.AddParameter(new StorageParameter
+						{
+							Value = GetValue(property),
+							Name = parameter.Name,
+							Type = parameter.Type,
+							Direction = parameter.Direction
+						});
+					}
+				}
+			}
+
+			return result;
+		}
+
+		Schema = Entity.GetSchemaAttribute();
+
+		return OnBuild();
+	}
+
+	protected List<PropertyInfo> Properties => _properties ??= GetProperties();
+
+	protected List<PropertyInfo> WhereProperties => _whereProperties;
+
+	protected string CommandText => Text.ToString();
+
+	protected IEntity Entity { get; private set; }
+
+	protected SchemaAttribute Schema { get; private set; }
+
+	private StringBuilder Text { get; set; }
+
+	protected abstract StorageOperation OnBuild();
+
+	protected abstract bool TryGetExisting(out StorageOperation? result);
+
+	protected List<IStorageParameter> Parameters => _parameters;
+
+	protected void Write(string text)
+	{
+		Text.Append(text);
+	}
+
+	protected void Write(char text)
+	{
+		Text.Append(text);
+	}
+
+	protected void WriteLine(string text)
+	{
+		Text.AppendLine(text);
+	}
+
+	protected void Trim()
+	{
+		for (var i = Text.Length - 1; i >= 0; i--)
+		{
+			if (!Text[i].Equals(',') && !Text[i].Equals('\n') && !Text[i].Equals('\r') && !Text[i].Equals(' '))
+				break;
+
+			if (i < Text.Length)
+				Text.Length = i;
+		}
+	}
+
+	protected virtual List<PropertyInfo> GetProperties()
+	{
+		return Interop.Properties.GetImplementedProperties(Entity);
+	}
+
+	protected static string ColumnName(PropertyInfo property)
+	{
+		var dataMember = property.FindAttribute<MemberAttribute>();
+
+		return dataMember is null || string.IsNullOrEmpty(dataMember.Member) ? property.Name.ToCamelCase() : dataMember.Member;
+	}
+
+	protected static DbType ResolveDbType(PropertyInfo property)
+	{
+		return property.PropertyType.ToDbType();
+	}
+
+	protected object? GetValue(PropertyInfo property)
+	{
+		if (IsNull(property))
+			return "NULL";
+
+		return GetValue(property.GetValue(Entity), property.PropertyType.ToDbType());
+	}
+
+	private static object? GetValue(object value, DbType dbType)
+	{
+		switch (dbType)
+		{
+			case DbType.Binary:
+				if (value is byte[] bytes)
+					return Convert.ToBase64String(bytes);
+				else
+					return Convert.ToBase64String(Encoding.UTF8.GetBytes(value.ToString()));
+			default:
+				return value;
+		}
+	}
+
+	private bool IsNull(PropertyInfo property)
+	{
+		var result = property.GetValue(Entity);
+
+		if (result is null)
+			return true;
+
+		if (property.GetCustomAttribute<NullableAttribute>() is null)
+			return false;
+
+		var def = Types.GetDefault(property.PropertyType);
+
+		return TypeComparer.Compare(result, def);
+	}
+
+	protected StorageParameter CreateParameter(PropertyInfo property)
+	{
+		return CreateParameter(property, ParameterDirection.Input);
+	}
+
+	protected StorageParameter CreateParameter(PropertyInfo property, ParameterDirection direction)
+	{
+		var columnName = ColumnName(property);
+		var parameterName = $"@{columnName}";
+
+		var parameter = new StorageParameter
+		{
+			Direction = direction,
+			Name = parameterName,
+			Type = ResolveDbType(property),
+			Value = GetValue(property)
+		};
+
+		Parameters.Add(parameter);
+
+		return parameter;
+	}
+
+	private PropertyInfo ResolveProperty(string parameterName)
+	{
+		var propertyName = parameterName[1..];
+		var flags = BindingFlags.Public | BindingFlags.Instance | BindingFlags.NonPublic;
+
+		if (Entity.GetType().GetProperty(propertyName.ToPascalCase(), flags) is PropertyInfo property)
+			return property;
+
+		if (Entity.GetType().GetProperty(propertyName, flags) is PropertyInfo raw)
+			return raw;
+
+		return null;
+	}
+}

Connected.ServiceModel.Client.Data/CqlFormatter.cs

@@ -1,8 +1,8 @@
-using Connected.Expressions;
+using System.Linq.Expressions;
+using Connected.Expressions;
 using Connected.Expressions.Formatters;
 using Connected.Expressions.Languages;
 using Connected.Interop;
-using System.Linq.Expressions;
 
 namespace Connected.ServiceModel.Client.Data;
 /// <summary>
@@ -14,7 +14,12 @@ internal sealed class CqlFormatter : SqlFormatter
 		  : base(language)
 	{
 		Context = context;
+
+		HideColumnAliases = true;
+		HideTableAliases = true;
+		UseBracketsInWhere = false;
 	}
+
 	public ExpressionCompilationContext Context { get; }
 
 	public static new string Format(ExpressionCompilationContext context, Expression expression)
@@ -36,6 +41,11 @@ internal sealed class CqlFormatter : SqlFormatter
 		else
 			base.WriteAggregateName(aggregateName);
 	}
+
+	protected override void WriteTableName(string tableSchema, string tableName)
+	{
+		Write(tableName);
+	}
 	protected override Expression VisitMemberAccess(MemberExpression m)
 	{
 		if (m.Member.DeclaringType == typeof(string))

Connected.ServiceModel.Client.Data/CqlLanguage.cs

@@ -37,12 +37,7 @@ internal sealed class CqlLanguage : QueryLanguage
 
 	public override string Quote(string name)
 	{
-		if (name.StartsWith("[") && name.EndsWith("]"))
 		return name;
-		else if (name.Contains('.'))
-			return $"[{string.Join("].[", name.Split(SplitChars, StringSplitOptions.RemoveEmptyEntries))}]";
-		else
-			return $"[{name}]";
 	}
 
 	public override Linguist CreateLinguist(ExpressionCompilationContext context, Translator translator)

Connected.ServiceModel.Client.Data/CqlLinguist.cs

@@ -1,8 +1,8 @@
-using Connected.Expressions;
+using System.Linq.Expressions;
+using Connected.Expressions;
 using Connected.Expressions.Languages;
 using Connected.Expressions.Translation;
 using Connected.Expressions.Translation.Rewriters;
-using System.Linq.Expressions;
 
 namespace Connected.ServiceModel.Client.Data;
 

Connected.ServiceModel.Client.Data/DataParameterCollection.cs

@@ -0,0 +1,61 @@
+using System.Data;
+
+namespace Connected.ServiceModel.Client.Data;
+internal sealed class DataParameterCollection : List<IDbDataParameter>, IDataParameterCollection
+{
+	public object this[string parameterName]
+	{
+		get
+		{
+			foreach (var parameter in this)
+			{
+				if (string.Equals(parameter.ParameterName, parameterName, StringComparison.OrdinalIgnoreCase))
+					return parameter;
+			}
+
+			throw new NullReferenceException(nameof(IDbDataParameter));
+		}
+		set
+		{
+			if (value is not IDbDataParameter parameter)
+				throw new InvalidCastException(nameof(IDbDataParameter));
+
+			var idx = IndexOf(parameterName);
+
+			if (idx < 0)
+				throw new NullReferenceException(nameof(IDbDataParameter));
+
+			this[idx] = parameter;
+		}
+	}
+
+	public bool Contains(string parameterName)
+	{
+		foreach (var parameter in this)
+		{
+			if (string.Equals(parameter.ParameterName, parameterName, StringComparison.OrdinalIgnoreCase))
+				return true;
+		}
+
+		return false;
+	}
+
+	public int IndexOf(string parameterName)
+	{
+		for (var i = 0; i < Count; i++)
+		{
+			var current = this[i];
+
+			if (string.Equals(current.ParameterName, parameterName, StringComparison.OrdinalIgnoreCase))
+				return i;
+		}
+
+		return -1;
+	}
+
+	public void RemoveAt(string parameterName)
+	{
+		if (this[parameterName] is IDbDataParameter target)
+			Remove(target);
+	}
+}

Connected.ServiceModel.Client.Data/DeleteCommandBuilder.cs

@@ -0,0 +1,56 @@
+using System.Collections.Concurrent;
+using System.Reflection;
+using Connected.Entities.Annotations;
+using Connected.Entities.Storage;
+
+namespace Connected.ServiceModel.Client.Data;
+internal sealed class DeleteCommandBuilder : CommandBuilder
+{
+	private static readonly ConcurrentDictionary<string, StorageOperation> _cache;
+
+	static DeleteCommandBuilder()
+	{
+		_cache = new();
+	}
+
+	private static ConcurrentDictionary<string, StorageOperation> Cache => _cache;
+
+	protected override StorageOperation OnBuild()
+	{
+		WriteLine($"DELETE [{Schema.Schema}].[{Schema.Name}] (");
+		WriteWhere();
+
+		var result = new StorageOperation { CommandText = CommandText };
+
+		foreach (var parameter in Parameters)
+			result.AddParameter(parameter);
+
+		Cache.TryAdd(Entity.GetType().FullName, result);
+
+		return result;
+	}
+
+	private void WriteWhere()
+	{
+		Write("WHERE ");
+
+		foreach (var property in Properties)
+		{
+			if (property.GetCustomAttribute<PrimaryKeyAttribute>() is not null)
+			{
+				var columnName = ColumnName(property);
+
+				CreateParameter(property);
+
+				Write($"{ColumnName} = @{ColumnName}");
+			}
+		}
+
+		Write(";");
+	}
+
+	protected override bool TryGetExisting(out StorageOperation? result)
+	{
+		return Cache.TryGetValue(Entity.GetType().FullName, out result);
+	}
+}

Connected.ServiceModel.Client.Data/InsertCommandBuilder.cs

@@ -0,0 +1,67 @@
+using System.Collections.Concurrent;
+using System.Reflection;
+using Connected.Entities.Annotations;
+using Connected.Entities.Storage;
+
+namespace Connected.ServiceModel.Client.Data;
+internal sealed class InsertCommandBuilder : CommandBuilder
+{
+	private static readonly ConcurrentDictionary<string, StorageOperation> _cache;
+
+	static InsertCommandBuilder()
+	{
+		_cache = new();
+	}
+
+	private static ConcurrentDictionary<string, StorageOperation> Cache => _cache;
+
+	protected override StorageOperation OnBuild()
+	{
+		WriteLine($"INSERT [{Schema.Schema}].[{Schema.Name}] (");
+
+		WriteColumns();
+		WriteLine(")");
+		Write("VALUES (");
+		WriteValues();
+		WriteLine(");");
+
+		var result = new StorageOperation { CommandText = CommandText };
+
+		foreach (var parameter in Parameters)
+			result.AddParameter(parameter);
+
+		Cache.TryAdd(Entity.GetType().FullName, result);
+
+		return result;
+	}
+
+	private void WriteColumns()
+	{
+		foreach (var property in Properties)
+		{
+			CreateParameter(property);
+
+			Write($"{ColumnName(property)}, ");
+		}
+
+		Trim();
+	}
+
+	private void WriteValues()
+	{
+		foreach (var property in Properties)
+		{
+			if (property.GetCustomAttribute<PrimaryKeyAttribute>() is not null)
+				continue;
+
+			Write($"@{ColumnName(property)}, ");
+		}
+
+		Trim();
+	}
+
+	protected override bool TryGetExisting(out StorageOperation? result)
+	{
+		return Cache.TryGetValue(Entity.GetType().FullName, out result);
+	}
+}

Connected.ServiceModel.Client.Data/Remote/RemoteTableColumn.cs

@@ -0,0 +1,12 @@
+using System.ComponentModel.DataAnnotations;
+
+namespace Connected.ServiceModel.Client.Data.Remote;
+internal sealed class RemoteTableColumn
+{
+	[Required, MaxLength(128)]
+	public string Name { get; set; } = default!;
+	[Required, MaxLength(128)]
+	public string DataType { get; set; } = default!;
+	public bool IsPartitionKey { get; set; }
+	public bool IsPrimaryKey { get; set; }
+}

Connected.ServiceModel.Client.Data/Remote/RemoteTableService.cs

@@ -0,0 +1,112 @@
+using System.Collections.Immutable;
+using System.Text.Json.Nodes;
+using Connected.Net;
+using Connected.ServiceModel.Client.Net;
+using Connected.ServiceModel.Client.Subscription;
+
+namespace Connected.ServiceModel.Client.Data.Remote;
+
+internal sealed class RemoteTableService
+{
+	//TODO:load it from config or environment.
+	private const string AccessToken = "Temp";
+	public RemoteTableService(IHttpService http, IConnectedServer server, ISubscriptionService subscription)
+	{
+		Http = http;
+		Server = server;
+		Subscription = subscription;
+	}
+
+	private IHttpService Http { get; }
+	private IConnectedServer Server { get; }
+	private ISubscriptionService Subscription { get; }
+	private int SubscriptionId { get; set; }
+
+	public async Task<JsonArray> Query(string commandText)
+	{
+		await Initialize();
+
+		var url = await Server.SelectUrl(new ConnectedServerUrlArgs { Kind = ConnectedUrlKind.TableStorage });
+
+		if (await Http.Post<JsonArray>($"{url}/query", new QueryTableArgs
+		{
+			CommandText = commandText,
+			AccessToken = AccessToken,
+			Subscription = SubscriptionId
+		}) is not JsonArray result)
+			return new JsonArray();
+
+		return result;
+	}
+
+	public async Task<ImmutableList<RemoteTableColumn>> QueryColumns(string tableName)
+	{
+		await Initialize();
+
+		var url = await Server.SelectUrl(new ConnectedServerUrlArgs { Kind = ConnectedUrlKind.TableStorage });
+
+		if (await Http.Post<List<RemoteTableColumn>>($"{url}/queryColumns", new TableSchemaArgs
+		{
+			TableName = tableName,
+			AccessToken = AccessToken,
+			Subscription = SubscriptionId
+		}) is not List<RemoteTableColumn> result)
+			return ImmutableList<RemoteTableColumn>.Empty;
+
+		return result.ToImmutableList();
+	}
+
+	public async Task<bool> TableExists(string tableName)
+	{
+		await Initialize();
+
+		var url = await Server.SelectUrl(new ConnectedServerUrlArgs { Kind = ConnectedUrlKind.TableStorage });
+
+		return await Http.Post<bool>($"{url}/tableExists", new TableSchemaArgs
+		{
+			TableName = tableName,
+			AccessToken = AccessToken,
+			Subscription = SubscriptionId
+		});
+	}
+
+	public async Task Update(string commandText)
+	{
+		await Initialize();
+
+		var url = await Server.SelectUrl(new ConnectedServerUrlArgs { Kind = ConnectedUrlKind.TableStorage });
+
+		await Http.Post($"{url}/update", new UpdateTableArgs
+		{
+			CommandText = commandText,
+			AccessToken = AccessToken,
+			Subscription = SubscriptionId
+		});
+	}
+
+	public async Task CreateTable(string tableName, List<RemoteTableColumn> columns)
+	{
+		await Initialize();
+
+		var url = await Server.SelectUrl(new ConnectedServerUrlArgs { Kind = ConnectedUrlKind.TableStorage });
+
+		await Http.Post($"{url}/createTable", new CreateTableArgs
+		{
+			Name = tableName,
+			Columns = columns,
+			AccessToken = AccessToken,
+			Subscription = SubscriptionId
+		});
+	}
+
+	private async Task Initialize()
+	{
+		if (SubscriptionId > 0)
+			return;
+
+		if (await Subscription.Select() is not ISubscription subscription)
+			throw new NullReferenceException(nameof(ISubscription));
+
+		SubscriptionId = subscription.Id;
+	}
+}

Connected.ServiceModel.Client.Data/Remote/TableArgs.cs

@@ -0,0 +1,33 @@
+using System.ComponentModel.DataAnnotations;
+using Connected.Annotations;
+
+namespace Connected.ServiceModel.Client.Data.Remote;
+internal class TableArgs : ServiceArgs
+{
+}
+
+internal sealed class QueryTableArgs : TableArgs
+{
+	[Required]
+	public string CommandText { get; set; } = default!;
+}
+
+internal sealed class TableSchemaArgs : ServiceArgs
+{
+	[Required, MaxLength(128)]
+	public string TableName { get; set; } = default!;
+}
+
+internal sealed class CreateTableArgs : ServiceArgs
+{
+	[Required, MaxLength(128)]
+	public string Name { get; set; } = default!;
+	[NonDefault]
+	public List<RemoteTableColumn> Columns { get; set; }
+}
+
+internal sealed class UpdateTableArgs : ServiceArgs
+{
+	[Required]
+	public string CommandText { get; set; } = default!;
+}

Connected.ServiceModel.Client.Data/Schema/ExistingSchema.cs

@@ -0,0 +1,45 @@
+using Connected.Data.Schema;
+
+namespace Connected.ServiceModel.Client.Data.Schema;
+internal sealed class ExistingSchema : ISchema
+{
+	public ExistingSchema()
+	{
+		Columns = new();
+	}
+
+	public List<ISchemaColumn> Columns { get; }
+
+	public string? Schema => null;
+
+	public string? Name { get; set; }
+
+	public string? Type { get; set; }
+
+	public bool Ignore { get; set; }
+
+	public async Task Load(SchemaExecutionContext context)
+	{
+		Name = context.Schema.Name;
+		Type = context.Schema.Type;
+
+		if (!await context.Remote.TableExists(context.Schema.Name))
+			return;
+
+		var columns = await context.Remote.QueryColumns(context.Schema.Name);
+
+		foreach (var column in columns)
+		{
+			Columns.Add(new SchemaColumn
+			{
+				Name = column.Name,
+				//TODO:populate properties
+			});
+		}
+	}
+
+	public bool Equals(ISchema? other)
+	{
+		throw new NotImplementedException();
+	}
+}

Connected.ServiceModel.Client.Data/Schema/SchemaColumn.cs

@@ -0,0 +1,43 @@
+using System.Data;
+using System.Reflection;
+using Connected.Data.Schema;
+using Connected.Entities.Annotations;
+
+namespace Connected.ServiceModel.Client.Data.Schema;
+internal sealed class SchemaColumn : ISchemaColumn
+{
+	public string? Name { get; set; }
+
+	public DbType DataType { get; set; }
+
+	public bool IsIdentity { get; set; }
+
+	public bool IsUnique { get; set; }
+
+	public bool IsIndex { get; set; }
+
+	public bool IsPrimaryKey { get; set; }
+
+	public bool IsVersion { get; set; }
+
+	public string? DefaultValue { get; set; }
+
+	public int MaxLength { get; set; }
+
+	public bool IsNullable { get; set; }
+
+	public string? Index { get; set; }
+
+	public int Scale { get; set; }
+
+	public int Precision { get; set; }
+
+	public DateKind DateKind { get; set; }
+
+	public BinaryKind BinaryKind { get; set; }
+
+	public int DatePrecision { get; set; }
+
+	public bool IsPartitionKey { get; set; }
+	public PropertyInfo Property { get; set; }
+}

Connected.ServiceModel.Client.Data/Schema/SchemaExecutionContext.cs

@@ -0,0 +1,16 @@
+using Connected.Data.Schema;
+using Connected.ServiceModel.Client.Data.Remote;
+
+namespace Connected.ServiceModel.Client.Data.Schema;
+internal sealed class SchemaExecutionContext
+{
+	public SchemaExecutionContext(ISchema schema, RemoteTableService remote)
+	{
+		Schema = schema;
+		Remote = remote;
+	}
+
+	public ExistingSchema ExistingSchema { get; set; }
+	public ISchema Schema { get; }
+	public RemoteTableService Remote { get; }
+}

Connected.ServiceModel.Client.Data/Schema/SynchronizationCommand.cs

@@ -0,0 +1,4 @@
+namespace Connected.ServiceModel.Client.Data.Schema;
+internal class SynchronizationCommand
+{
+}

Connected.ServiceModel.Client.Data/Schema/SynchronizationQuery.cs

@@ -0,0 +1,19 @@
+namespace Connected.ServiceModel.Client.Data.Schema;
+internal class SynchronizationQuery<T> : SynchronizationCommand
+{
+	protected SchemaExecutionContext Context { get; private set; }
+
+	public async Task<T> Execute(SchemaExecutionContext context)
+	{
+		Context = context;
+
+		return await OnExecute();
+	}
+
+	protected virtual async Task<T> OnExecute()
+	{
+		await Task.CompletedTask;
+
+		return default;
+	}
+}

Connected.ServiceModel.Client.Data/Schema/SynchronizationTransaction.cs

@@ -0,0 +1,17 @@
+namespace Connected.ServiceModel.Client.Data.Schema;
+internal class SynchronizationTransaction : SynchronizationCommand
+{
+	protected SchemaExecutionContext Context { get; private set; }
+
+	public async Task Execute(SchemaExecutionContext context)
+	{
+		Context = context;
+
+		await OnExecute();
+	}
+
+	protected virtual async Task OnExecute()
+	{
+		await Task.CompletedTask;
+	}
+}

Connected.ServiceModel.Client.Data/Schema/TableCreate.cs

@@ -0,0 +1,127 @@
+using Connected.Collections;
+using Connected.Interop;
+using Connected.ServiceModel.Annotations;
+using Connected.ServiceModel.Client.Data.Remote;
+
+namespace Connected.ServiceModel.Client.Data.Schema;
+internal sealed class TableCreate : TableTransaction
+{
+	//private const string KeysExceptionMessage = "A ISchemaColumn cannot have a PartitionKeyAttribute and PrimaryKeyAttribute set. Use either PartitionKey or PrimaryKey attribute.";
+	//private const string NoPartitionKeyMessage = "Schema must have at least one property with PartitionKeyAttribute.";
+
+	protected override async Task OnExecute()
+	{
+		var columns = new List<RemoteTableColumn>();
+
+		Context.Schema.Columns.SortByOrdinal();
+
+		foreach (var column in Context.Schema.Columns)
+		{
+			columns.Add(new RemoteTableColumn
+			{
+				DataType = CreateDataTypeMetaData(column),
+				IsPartitionKey = column.Property?.FindAttribute<PartitionKeyAttribute>() is not null,
+				IsPrimaryKey = column.IsPrimaryKey,
+				Name = column.Name
+			});
+		}
+
+		await Context.Remote.CreateTable(Context.Schema.Name, columns);
+	}
+
+	//private List<ISchemaColumn> PartitionKeys
+	//{
+	//	get
+	//	{
+	//		var result = new List<ISchemaColumn>();
+
+	//		foreach (var column in Context.Schema.Columns)
+	//		{
+	//			if (column.Property?.FindAttribute<PartitionKeyAttribute>() is not null)
+	//			{
+	//				if (column.IsPrimaryKey)
+	//					throw new InvalidOperationException($"{KeysExceptionMessage} ({column.Name})");
+
+	//				result.Add(column);
+	//			}
+	//		}
+
+	//		return result;
+	//	}
+	//}
+
+	//private List<ISchemaColumn> PrimaryKeys
+	//{
+	//	get
+	//	{
+	//		var result = new List<ISchemaColumn>();
+
+	//		foreach (var column in Context.Schema.Columns)
+	//		{
+	//			if (column.IsPrimaryKey)
+	//			{
+	//				if (column.Property?.FindAttribute<PartitionKeyAttribute>() is not null)
+	//					throw new InvalidOperationException($"{KeysExceptionMessage} ({column.Name})");
+
+	//				result.Add(column);
+	//			}
+	//		}
+
+	//		return result;
+	//	}
+	//}
+
+	//private string CommandText
+	//{
+	//	get
+	//	{
+	//		var partitionKeys = PartitionKeys;
+	//		var primaryKeys = PrimaryKeys;
+
+	//		if (!partitionKeys.Any())
+	//			throw new InvalidOperationException($"{NoPartitionKeyMessage} ({Context.Schema.Name})");
+
+	//		var keysCount = partitionKeys.Count + primaryKeys.Count;
+	//		var text = new StringBuilder();
+	//		var name = Temporary ? TemporaryName : Context.Schema.Name;
+
+	//		text.AppendLine($"CREATE TABLE {name}");
+	//		text.AppendLine("(");
+	//		var comma = string.Empty;
+
+	//		for (var i = 0; i < Context.Schema.Columns.Count; i++)
+	//		{
+	//			text.AppendLine($"{comma} {CreateColumnCommandText(Context.Schema.Columns[i])}");
+
+	//			comma = ",";
+	//		}
+
+	//		text.Append("PRIMARY KEY (");
+
+	//		if (keysCount > 1)
+	//			text.Append('(');
+
+	//		for (var i = 0; i < partitionKeys.Count; i++)
+	//		{
+	//			text.Append(partitionKeys[i].Name);
+
+	//			if (i < partitionKeys.Count - 1)
+	//				text.Append(',');
+	//		}
+
+	//		if (keysCount > 1)
+	//			text.Append(')');
+
+	//		for (var i = 0; i < primaryKeys.Count; i++)
+	//		{
+	//			text.Append(primaryKeys[i].Name);
+
+	//			if (i < primaryKeys.Count - 1)
+	//				text.Append(',');
+	//		}
+
+	//		text.AppendLine(");");
+
+	//		return text.ToString();
+	//	}
+}

Connected.ServiceModel.Client.Data/Schema/TableExists.cs

@@ -0,0 +1,8 @@
+namespace Connected.ServiceModel.Client.Data.Schema;
+internal class TableExists : SynchronizationQuery<bool>
+{
+	protected override async Task<bool> OnExecute()
+	{
+		return await Context.Remote.TableExists(Context.Schema.Name);
+	}
+}

Connected.ServiceModel.Client.Data/Schema/TableSynchronize.cs

@@ -0,0 +1,93 @@
+using Connected.Data.Schema;
+
+namespace Connected.ServiceModel.Client.Data.Schema;
+internal sealed class TableSynchronize : TableTransaction
+{
+	private ExistingSchema? _existingSchema;
+
+	private bool TableExists { get; set; }
+
+	protected override async Task OnExecute()
+	{
+		TableExists = await new TableExists().Execute(Context);
+
+		if (!TableExists)
+		{
+			await new TableCreate().Execute(Context);
+			return;
+		}
+
+		return;
+
+		_existingSchema = new();
+
+		await _existingSchema.Load(Context);
+
+		Context.ExistingSchema = ExistingSchema;
+
+		//TODO: implement alter table
+
+		//if (ShouldRecreate)
+		//	await new TableRecreate(ExistingSchema).Execute(Context);
+		//else if (ShouldAlter)
+		//	await new TableAlter(ExistingSchema).Execute(Context);
+	}
+
+	private bool ShouldAlter => !Context.Schema.Equals(ExistingSchema);
+	private bool ShouldRecreate => HasIdentityChanged || HasColumnMetadataChanged;
+
+	private ExistingSchema? ExistingSchema => _existingSchema;
+
+	private bool HasIdentityChanged
+	{
+		get
+		{
+			foreach (var column in Context.Schema.Columns)
+			{
+				if (ExistingSchema.Columns.FirstOrDefault(f => string.Equals(f.Name, column.Name, StringComparison.OrdinalIgnoreCase)) is not ISchemaColumn existing)
+					return true;
+
+				if (existing.IsIdentity != column.IsIdentity)
+					return true;
+			}
+
+			foreach (var existing in ExistingSchema.Columns)
+			{
+				var column = Context.Schema.Columns.FirstOrDefault(f => string.Equals(f.Name, existing.Name, StringComparison.OrdinalIgnoreCase));
+
+				if (column is null && existing.IsIdentity)
+					return true;
+				else if (column is not null && column.IsIdentity != existing.IsIdentity)
+					return true;
+			}
+
+			return false;
+		}
+	}
+
+	private bool HasColumnMetadataChanged
+	{
+		get
+		{
+			foreach (var existing in ExistingSchema.Columns)
+			{
+				if (Context.Schema.Columns.FirstOrDefault(f => string.Equals(f.Name, existing.Name, StringComparison.OrdinalIgnoreCase)) is not ISchemaColumn column)
+					continue;
+
+				if (column.DataType != existing.DataType
+					|| column.MaxLength != existing.MaxLength
+					|| column.IsNullable != existing.IsNullable
+					|| column.IsVersion != existing.IsVersion
+					|| column.Precision != existing.Precision
+					|| column.Scale != existing.Scale
+					|| column.DateKind != existing.DateKind
+					|| column.BinaryKind != existing.BinaryKind
+					|| column.DatePrecision != existing.DatePrecision)
+					return true;
+			}
+
+			return false;
+		}
+	}
+}
+

Connected.ServiceModel.Client.Data/Schema/TableTransaction.cs

@@ -0,0 +1,51 @@
+using System.Data;
+using System.Text;
+using Connected.Data.Schema;
+
+namespace Connected.ServiceModel.Client.Data.Schema;
+internal class TableTransaction : SynchronizationTransaction
+{
+	protected static string CreateColumnCommandText(ISchemaColumn column)
+	{
+		var builder = new StringBuilder();
+
+		builder.AppendFormat($"{column.Name} {CreateDataTypeMetaData(column)} ");
+
+		return builder.ToString();
+	}
+
+	protected static string CreateDataTypeMetaData(ISchemaColumn column)
+	{
+		return column.DataType switch
+		{
+			DbType.AnsiString => "text",
+			DbType.Binary => "blob",
+			DbType.Byte => "tinyint",
+			DbType.Boolean => "boolean",
+			DbType.Currency => "decimal",
+			DbType.Date => "timestamp",
+			DbType.DateTime => "timestamp",
+			DbType.Decimal => "decimal",
+			DbType.Double => "double",
+			DbType.Guid => "uuid",
+			DbType.Int16 => "smallint",
+			DbType.Int32 => "int",
+			DbType.Int64 => "bigint",
+			DbType.Object => "blob",
+			DbType.SByte => "smallint",
+			DbType.Single => "float",
+			DbType.String => "text",
+			DbType.Time => "time",
+			DbType.UInt16 => "int",
+			DbType.UInt32 => "bigint",
+			DbType.UInt64 => "float",
+			DbType.VarNumeric => "decimal",
+			DbType.AnsiStringFixedLength => "text",
+			DbType.StringFixedLength => "text",
+			DbType.Xml => "text",
+			DbType.DateTime2 => "timestamp",
+			DbType.DateTimeOffset => "timestamp",
+			_ => throw new NotSupportedException(),
+		};
+	}
+}

Connected.ServiceModel.Client.Data/TableConnection.cs

@@ -1,4 +1,5 @@
 using System.Data;
+using Connected.ServiceModel.Client.Data.Remote;
 
 namespace Connected.ServiceModel.Client.Data;
 /// <summary>
@@ -6,6 +7,14 @@ namespace Connected.ServiceModel.Client.Data;
 /// </summary>
 internal sealed class TableConnection : IDbConnection
 {
+	public TableConnection(RemoteTableService tables, string connectionString)
+	{
+		Tables = tables;
+		ConnectionString = connectionString;
+	}
+
+	public RemoteTableService Tables { get; }
+
 	/// <summary>
 	/// The connection string (URL) used when performing the requests. 
 	/// </summary>
@@ -60,15 +69,14 @@ internal sealed class TableConnection : IDbConnection
 
 	public IDbCommand CreateCommand()
 	{
-		throw new NotImplementedException();
+		return new TableDataCommand(Tables);
 	}
 
 	public void Dispose()
 	{
-		throw new NotImplementedException();
 	}
 	/// <summary>
-	/// This method doesn't do enything since the connection is stateless any is theoretically
+	/// This method doesn't do anything since the connection is stateless any is theoretically
 	/// always in open state.
 	/// </summary>
 	public void Open()

Connected.ServiceModel.Client.Data/TableDataCommand.cs

@@ -0,0 +1,64 @@
+using System.Data;
+using Connected.ServiceModel.Client.Data.Remote;
+
+namespace Connected.ServiceModel.Client.Data;
+internal sealed class TableDataCommand : IDbCommand
+{
+	public TableDataCommand(RemoteTableService tables)
+	{
+		Parameters = new DataParameterCollection();
+		Tables = tables;
+	}
+	public string CommandText { get; set; }
+	public int CommandTimeout { get; set; }
+	public CommandType CommandType { get; set; }
+	public IDbConnection? Connection { get; set; }
+
+	public IDataParameterCollection Parameters { get; }
+
+	public IDbTransaction? Transaction { get; set; }
+	public UpdateRowSource UpdatedRowSource { get; set; }
+	public RemoteTableService Tables { get; }
+
+	public void Cancel()
+	{
+
+	}
+
+	public IDbDataParameter CreateParameter()
+	{
+		return new TableDataParameter();
+	}
+
+	public void Dispose()
+	{
+
+	}
+
+	public int ExecuteNonQuery()
+	{
+		AsyncUtils.RunSync(() => Tables.Update(CommandText));
+
+		return 0;
+	}
+
+	public IDataReader ExecuteReader()
+	{
+		return new TableDataReader(AsyncUtils.RunSync(() => Tables.Query(CommandText)));
+	}
+
+	public IDataReader ExecuteReader(CommandBehavior behavior)
+	{
+		return new TableDataReader(AsyncUtils.RunSync(() => Tables.Query(CommandText)));
+	}
+
+	public object? ExecuteScalar()
+	{
+		throw new NotImplementedException();
+	}
+
+	public void Prepare()
+	{
+		throw new NotImplementedException();
+	}
+}

Connected.ServiceModel.Client.Data/TableDataConnection.cs

@@ -1,16 +1,19 @@
-using Connected.Annotations;
+using System.Data;
+using Connected.Annotations;
 using Connected.Data.Storage;
+using Connected.ServiceModel.Client.Data.Remote;
 using Microsoft.Data.SqlClient;
-using System.Data;
 
 namespace Connected.ServiceModel.Client.Data;
 
 [ServiceRegistration(ServiceRegistrationMode.Auto, ServiceRegistrationScope.Transient)]
 internal sealed class TableDataConnection : DatabaseConnection
 {
-	public TableDataConnection(ICancellationContext context) : base(context)
+	public TableDataConnection(RemoteTableService tables, ICancellationContext context) : base(context)
 	{
+		Tables = tables;
 	}
+	private RemoteTableService Tables { get; }
 
 	protected override void SetupParameters(IStorageCommand command, IDbCommand cmd)
 	{
@@ -54,6 +57,6 @@ internal sealed class TableDataConnection : DatabaseConnection
 	{
 		await Task.CompletedTask;
 
-		return new SqlConnection(ConnectionString);
+		return new TableConnection(Tables, ConnectionString);
 	}
 }

Connected.ServiceModel.Client.Data/TableDataParameter.cs

@@ -0,0 +1,16 @@
+using System.Data;
+
+namespace Connected.ServiceModel.Client.Data;
+internal class TableDataParameter : IDbDataParameter
+{
+	public byte Precision { get; set; }
+	public byte Scale { get; set; }
+	public int Size { get; set; }
+	public DbType DbType { get; set; }
+	public ParameterDirection Direction { get; set; }
+	public bool IsNullable { get; set; }
+	public string ParameterName { get; set; }
+	public string SourceColumn { get; set; }
+	public DataRowVersion SourceVersion { get; set; }
+	public object? Value { get; set; }
+}

Connected.ServiceModel.Client.Data/TableDataReader.cs

@@ -0,0 +1,164 @@
+using System.Data;
+using System.Diagnostics.CodeAnalysis;
+using System.Text.Json.Nodes;
+
+namespace Connected.ServiceModel.Client.Data;
+internal sealed class TableDataReader : IDataReader
+{
+	public TableDataReader(JsonArray items)
+	{
+		Items = items;
+	}
+
+	public object this[int i] => throw new NotImplementedException();
+
+	public object this[string name] => throw new NotImplementedException();
+
+	public int Depth => throw new NotImplementedException();
+
+	public bool IsClosed => throw new NotImplementedException();
+
+	public int RecordsAffected => throw new NotImplementedException();
+
+	public int FieldCount => throw new NotImplementedException();
+
+	public JsonArray Items { get; }
+
+	public void Close()
+	{
+		throw new NotImplementedException();
+	}
+
+	public void Dispose()
+	{
+		throw new NotImplementedException();
+	}
+
+	public bool GetBoolean(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public byte GetByte(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public long GetBytes(int i, long fieldOffset, byte[]? buffer, int bufferoffset, int length)
+	{
+		throw new NotImplementedException();
+	}
+
+	public char GetChar(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public long GetChars(int i, long fieldoffset, char[]? buffer, int bufferoffset, int length)
+	{
+		throw new NotImplementedException();
+	}
+
+	public IDataReader GetData(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public string GetDataTypeName(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public DateTime GetDateTime(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public decimal GetDecimal(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public double GetDouble(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	[return: DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicFields | DynamicallyAccessedMemberTypes.PublicProperties)]
+	public Type GetFieldType(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public float GetFloat(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public Guid GetGuid(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public short GetInt16(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public int GetInt32(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public long GetInt64(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public string GetName(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public int GetOrdinal(string name)
+	{
+		throw new NotImplementedException();
+	}
+
+	public DataTable? GetSchemaTable()
+	{
+		throw new NotImplementedException();
+	}
+
+	public string GetString(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public object GetValue(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public int GetValues(object[] values)
+	{
+		throw new NotImplementedException();
+	}
+
+	public bool IsDBNull(int i)
+	{
+		throw new NotImplementedException();
+	}
+
+	public bool NextResult()
+	{
+		throw new NotImplementedException();
+	}
+
+	public bool Read()
+	{
+		//var items = AsyncUtils.RunSync(() => Service.Query(CommandText));
+
+		return false;
+	}
+}

Connected.ServiceModel.Client.Data/TableSchemaMiddleware.cs

@@ -1,7 +1,11 @@
 using Connected.Annotations;
 using Connected.Data.Schema;
+using Connected.Entities.Annotations;
 using Connected.Entities.Storage;
+using Connected.Interop;
 using Connected.Middleware;
+using Connected.ServiceModel.Client.Data.Remote;
+using Connected.ServiceModel.Client.Data.Schema;
 using Connected.ServiceModel.Client.Net;
 using Connected.ServiceModel.Data;
 
@@ -10,13 +14,15 @@ namespace Connected.ServiceModel.Client.Data;
 [Priority(2)]
 internal sealed class TableSchemaMiddleware : MiddlewareComponent, ISchemaMiddleware
 {
-	public TableSchemaMiddleware(IMiddlewareService middleware, IStorageProvider storage, IConnectedServer connected)
+	public TableSchemaMiddleware(IMiddlewareService middleware, IStorageProvider storage, RemoteTableService remote, IConnectedServer connected)
 	{
 		Storage = storage;
+		Remote = remote;
 		Connected = connected;
 	}
 	private IStorageProvider Storage { get; }
-	public IConnectedServer Connected { get; }
+	private RemoteTableService Remote { get; }
+	private IConnectedServer Connected { get; }
 	public Type ConnectionType => typeof(TableDataConnection);
 	public string DefaultConnectionString { get; private set; } = default!;
 
@@ -31,7 +37,7 @@ internal sealed class TableSchemaMiddleware : MiddlewareComponent, ISchemaMiddle
 		/*
 		 * This middleware supports all ITableEntity<> entities.
 		 */
-		return entityType.IsAssignableTo(typeof(ITableEntity<,>));
+		return entityType.ImplementsInterface(typeof(ITableEntity<>));
 	}
 
 	public async Task Synchronize(Type entity, ISchema schema)
@@ -41,17 +47,9 @@ internal sealed class TableSchemaMiddleware : MiddlewareComponent, ISchemaMiddle
 
 	private async Task Synchronize(ISchema schema, string connectionString)
 	{
-		//var args = new SchemaExecutionContext(Storage, schema, connectionString);
-		///*
-		// * Sinchronize schema object first.
-		// */
-		//await new SchemaSynchronize().Execute(args);
-		///*
-		// * Only tables are supported
-		// */
-		//if (string.IsNullOrWhiteSpace(schema.Type) || string.Equals(schema.Type, SchemaAttribute.SchemaTypeTable, StringComparison.OrdinalIgnoreCase))
-		//	await new TableSynchronize().Execute(args);
+		var args = new SchemaExecutionContext(schema, Remote);
 
-		await Task.CompletedTask;
+		if (string.IsNullOrWhiteSpace(schema.Type) || string.Equals(schema.Type, SchemaAttribute.SchemaTypeTable, StringComparison.OrdinalIgnoreCase))
+			await new TableSynchronize().Execute(args);
 	}
 }

Connected.ServiceModel.Client.Data/TableStorageProvider.cs

@@ -9,6 +9,7 @@ using Connected.Expressions;
 using Connected.Expressions.Evaluation;
 using Connected.Expressions.Query;
 using Connected.Expressions.Translation;
+using Connected.Interop;
 using Connected.ServiceModel.Data;
 
 namespace Connected.ServiceModel.Client.Data;
@@ -82,19 +83,18 @@ internal class TableStorageProvider : QueryProvider, IStorageExecutor, IStorageM
 
 	public bool SupportsEntity(Type entityType)
 	{
-		return entityType.IsAssignableTo(typeof(ITableEntity<,>));
+		return entityType.ImplementsInterface(typeof(ITableEntity<>));
 	}
 
 	public IStorageOperation CreateOperation<TEntity>(TEntity entity)
 		where TEntity : IEntity
 	{
-		//var builder = new AggregatedCommandBuilder<TEntity?>();
+		var builder = new AggregatedCommandBuilder<TEntity?>();
 
-		//if (builder.Build(entity) is not StorageOperation operation)
-		//	throw new NullReferenceException(nameof(StorageOperation));
+		if (builder.Build(entity) is not StorageOperation operation)
+			throw new NullReferenceException(nameof(StorageOperation));
 
-		//return operation;
-		return null;
+		return operation;
 	}
 
 	public IStorageReader<TEntity> OpenReader<TEntity>(IStorageOperation operation, IStorageConnection connection)

Connected.ServiceModel.Client.Data/UpdateCommandBuilder.cs

@@ -0,0 +1,77 @@
+using System.Collections.Concurrent;
+using System.Reflection;
+using Connected.Entities.Annotations;
+using Connected.Entities.Storage;
+
+namespace Connected.ServiceModel.Client.Data;
+internal sealed class UpdateCommandBuilder : CommandBuilder
+{
+	private static readonly ConcurrentDictionary<string, StorageOperation> _cache;
+
+	static UpdateCommandBuilder()
+	{
+		_cache = new();
+	}
+
+	private static ConcurrentDictionary<string, StorageOperation> Cache => _cache;
+	private bool SupportsConcurrency { get; set; }
+	protected override bool TryGetExisting(out StorageOperation? result)
+	{
+		return Cache.TryGetValue(Entity.GetType().FullName, out result);
+	}
+
+	protected override StorageOperation OnBuild()
+	{
+		WriteLine($"UPDATE [{Schema.Schema}].[{Schema.Name}] SET");
+
+		WriteAssignments();
+		WriteWhere();
+
+		Trim();
+		Write(';');
+
+		var result = new StorageOperation { CommandText = CommandText, Concurrency = SupportsConcurrency ? DataConcurrencyMode.Enabled : DataConcurrencyMode.Disabled };
+
+		foreach (var parameter in Parameters)
+			result.AddParameter(parameter);
+
+		Cache.TryAdd(Entity.GetType().FullName, result);
+
+		return result;
+	}
+
+	private void WriteAssignments()
+	{
+		foreach (var property in Properties)
+		{
+			if (property.GetCustomAttribute<PrimaryKeyAttribute>() is not null)
+			{
+				WhereProperties.Add(property);
+
+				continue;
+			}
+
+			var parameter = CreateParameter(property);
+
+			WriteLine($"{ColumnName(property)} = {parameter.Name},");
+		}
+
+		Trim();
+	}
+
+	private void WriteWhere()
+	{
+		WriteLine(string.Empty);
+
+		for (var i = 0; i < WhereProperties.Count; i++)
+		{
+			var property = WhereProperties[i];
+			var parameter = CreateParameter(property);
+
+			if (i == 0)
+				WriteLine($" WHERE {ColumnName(property)} = {parameter.Name}");
+			else
+				WriteLine($" AND {ColumnName(property)} = {parameter.Name}");
+		}
+	}
+}

Connected.ServiceModel.Client/Net/ConnectedServerOps.cs

@@ -4,7 +4,7 @@ namespace Connected.ServiceModel.Client.Net;
 
 internal sealed class SelectUrl : ServiceFunction<ConnectedServerUrlArgs, string>
 {
-	private const string Root = "https://connected.tompit.com";
+	private const string Root = "https://localhost:61599";//"https://connected.tompit.com";
 	protected override async Task<string?> OnInvoke()
 	{
 		await Task.CompletedTask;