local cassandra = require "lua_cassandra"
local ES = require "fullcontact.elastic_search"
local utf8 = require "icu.utf8"
require "uuid"
require "LuaOrm.Date"
local pl = {
	tablex = require "pl.tablex",
	func = require "pl.func",
	operator = require "pl.operator"
}

local helpers = require "fullcontact.helpers"
local m_configValues = helpers.GetConfigValues()
local logger = require "fullcontact.logger"
local AddressMapper = require "fullcontact.address_mapper"
local json = require "cjson"
local stp = require "StackTracePlus"
-- Tope máximo de filas que se pueden importar en una importacion de dialer contacts.
local MAXIMUM_NUMBER_OF_ROWS_IN_DIALER_IMPORT = m_configValues.importation.maximumNumberOfRowsInDialerImport
local ROWS_PER_PAGE = m_configValues.importation.rowsPerPage
-- Tipo de importacion (SINGLE O BATCH) --> SINGLE is DEPRECATED.
local IMPORTATION_TYPE = m_configValues.importation.type

json.encode_empty_table_as_array(true)

local m_datasource = {}
local public = {}

-- iterator that returns only keys
local function keys (t)
	local k
	return function() k = next(t, k); return k end
end

-- iterator that returns only values
local function values (t)
	local k
	return function() k = next(t, k); return t[k] end
end

local function pack (...)
	return { n = select("#", ...), ... }
end

local function try (finally, ok, ...)
	finally()
	if ok then
		return ...
	end
	logger.error(...)
	return nil, ...
end

---
-- Returns a function that wraps the call to a given function providing a CassandraConnection bound to a given
-- keyspace or "Contacts" by default.
-- Either on success or error, the connection is closed.
--
local function with_connection (keyspace, f)
	local connection
	if type(keyspace) == "function" then
		f = keyspace
		connection = m_datasource.cassandra:CassandraConnection("Contacts")
	else
		connection = m_datasource.cassandra:CassandraConnection(keyspace)
	end

	local function finally()
		connection:Close()
	end

	return function(...)
		local args = pack(...)
		return try(
			finally, xpcall( function() return f(connection, unpack(args, 1, args.n)) end, stp.stacktrace )
		)
	end
end

---
-- Calls a given function providing a CassandraConnection bound to a given keyspace or "Contacts" by default.
-- Either on success or error, the connection is closed.
--
-- Could not implement this in terms of the other due to keyspace being optional.
--
local function with_connection_do (keyspace, f, ...)
	local connection, args

	if type(keyspace) == "function" then
		args = pack(f, ...)
		f = keyspace
		connection = m_datasource.cassandra:CassandraConnection("Contacts")
	else
		connection = m_datasource.cassandra:CassandraConnection(keyspace)
		args = pack(...)
	end

	local function finally()
		connection:Close()
	end

	return try(
		finally, xpcall( function() return f(connection, unpack(args, 1, args.n)) end, stp.stacktrace )
	)
end


---
-- given a contact, returns a table with the values that can be stored in Cassandra.
-- This just skips 'addresses', 'campaigns' (which are tables) and anything not one of
-- string, number or boolean.
--
local function get_values_to_store (contact)
	local values_to_store = pl.tablex.pairmap(function(col_name, col_value)
		local value_type = type(col_value)
		if col_name ~= "addresses" and col_name ~= "campaigns" and
			(value_type == "string" or value_type == "number" or value_type == "boolean")
		then
			return col_value, col_name
		end
	end, contact)
	--logger.debug("[FullContact::get_values_to_store] --> values_to_store '%s'", json.encode(values_to_store or {}))
	return values_to_store
end

local function build_address_to_index (addresses, vcc)
	for _, address in pairs(addresses or {}) do
		if address.Type then
			local address_type = utf8.lower(address.Type)
			-- from an address, get the data that needs to be written in Cassandra
			-- and the value to store in the Search Engine for this address
			local addressInfo = AddressMapper[address_type](address, vcc)
			if addressInfo then
				addresses[_] = addressInfo
			end
		else
			logger.info("[FullContact::build_address_to_index] NOT address.Type. address '%s'",json.encode(address))
		end
	end
	return addresses
end
---
-- Some metatables to ease manipulation

local Addresses_mt = {}
Addresses_mt.__index = Addresses_mt

function Addresses_mt:Add (address, vcc)
	local success, addressId = helpers.BuildAddressId(address.Type, address, vcc)
	-- if not success then
	-- 	local err = addressId
	-- 	logger.warn("[FullContact::Addresses_mt:Add] Unable to add address '%s'. Error '%s'", json.encode(address), json.encode(err))
	-- 	return
	-- end
	address.AccountId = address.AccountId or addressId
	local found
	for i, addr in ipairs(self) do
		if addr.AccountId == address.AccountId then
			self[i] = address
			found = true
			break
		end
	end
	if not found then
		table.insert(self, address)
	end
end

function Addresses_mt:Remove (address, vcc)
	local success, addressId = helpers.BuildAddressId(address.Type, address, vcc)
	-- if not success then
	-- 	local err = addressId
	-- 	logger.warn("[FullContact::Addresses_mt:Remove] Unable to remove address '%s'. Error '%s'", json.encode(address), json.encode(err))
	-- 	return
	-- end
	address.AccountId = address.AccountId or addressId
	local idx = pl.tablex.find_if(self, function(addr) return addr.AccountId == address.AccountId end)
	if idx then table.remove(self, idx) end
end

function Addresses_mt:Find (address, vcc)
	local success, addressId = helpers.BuildAddressId(address.Type, address, vcc)
	-- if not success then
	-- 	local err = addressId
	-- 	logger.warn("[FullContact::Addresses_mt:Find] Unable to find address '%s'. Error '%s'", json.encode(address), json.encode(err))
	-- 	return
	-- end
	local found
	for i, addr in ipairs(self) do
		if addr.AccountId == addressId then
			return true
		end
	end
	return false
end

local Campaigns_mt = {}
Campaigns_mt.__index = Campaigns_mt

function Campaigns_mt:Add (campaign_id)
	for _, campaign in ipairs(self) do
		if campaign == campaign_id then
			return
		end
	end
	table.insert(self, campaign_id)
end

function Campaigns_mt:Remove (campaign_id)
	local idx = pl.tablex.find_if(self, function(c) return c == campaign_id end)
	if idx then table.remove(self, idx) end
end

local Contact_mt = {}
Contact_mt.__index = Contact_mt

function Contact_mt:Update ()
	return public.Save(self)
end


---
--
local function get_contact (connection, contact_id, options)
	options = options or {}

	--logger.debug("[FULLCONTACT:get_contact] - contact_id '%s'", contact_id)

	local contact = setmetatable({}, Contact_mt)

	-- basic data
	local basicColumns = connection:GetSliceRange(contact_id,
		"Contacts",
		"",
		{
			column_names = {},
			slice_range = {
				start = "",
				finish = "",
				reversed = false,
				count = 200
			}
		}
	)
	if basicColumns then
		pl.tablex.foreach(basicColumns, function(v)
			-- treat __handles column as a special case
			if v.name:match("^__handles") then
				contact.handles = helpers.JSON_Decode(v.value)
			else
				contact[v.name] = v.value
			end
		end)
	end
	-- if no columns are projected this is a tombstone and should not be returned
	if not contact.Id then
		return
	end
	if contact.__logically_deleted and not options.logical then
		return
	end

	-- default value for handles
	contact.handles = contact.handles or {}

	-- contact addresses, hard limited at 100 addresses
	local rows = connection:GetSuperRangeSlices("ContactAddresses", --column family
		"",  --supercolumn
		contact_id, --startkey
		contact_id, --endkey
		{
			column_names = {},
			slice_range = {
				start = "",
				finish = "",
				reversed = false,
				count = 100
			}
		},
		100 --total keys
	) or {}

	contact.addresses = setmetatable({}, Addresses_mt)
	for key, contactAddresses in rows:Iterator() do
		if key == contact_id then
			--estoy en la que pedi, no deberia volver mas de un row anyway
			for v in values(contactAddresses) do
				local canIncludeAddress = true
				local address = {}
				--v es la supercolumn de cada entrada con la pareja name/subcolumns
				for colid, column in pairs(v.columns) do
					--logger.debug("[FULLCONTACT:get_contact]    column.name: '%s', column.value: %s", column.name, column.value)
					address[column.name] = helpers.JSON_Decode(column.value)
				end
				address.AccountId = v.name
				-- patch old contacts, add the new 'displayable' field
				if not address.displayable then
					local addressInfo = AddressMapper[utf8.lower(address.Type)](address, contact.VCC)
					if addressInfo then
						address.displayable = addressInfo.displayable
						connection:InsertSuperColumn(contact_id, "ContactAddresses", address.AccountId, "displayable", address.displayable)
					else
						canIncludeAddress = false
					end
				end
				if canIncludeAddress then
					table.insert(contact.addresses, address)
				end
			end
			break
		end
	end

	-- contact campaigns, hard limited at 200 campaigns
	contact.campaigns = setmetatable({}, Campaigns_mt)
	local campaignColumns = connection:GetSliceRange(contact_id,
		"ContactCampaigns",
		"",
		{
			column_names = {},
			slice_range = {
				start = "",
				finish = "",
				reversed = false,
				count = 200
			}
		}
	)
	if campaignColumns then
		pl.tablex.foreach(campaignColumns, function(column) table.insert(contact.campaigns, column.name) end)
	end

	--logger.debug("[FULLCONTACT:get_contact] ------> BY CASSANDRA '%s'", helpers.JSON_Encode(contact or {}))

	return contact
end

---
--	Finds a contact from a given address, now the contacts are filtered by VCC but the alternative to filter contacts by campaign/group
--	Is there too, that would change the way the contact keys are built when automatically creating contacts from twitter/facebook since only @vcc is appended
-- TODO: fix comment
--
local function get_contact_by_address (connection, addressType, address, vcc)
	--logger.debug("[FullContact::get_contact_by_address] 1")
	local success, addressId = helpers.BuildAddressId(addressType, address, vcc)
	-- if not success then
	-- 	local err = addressId
	-- 	logger.warn("[FullContact::get_contact_by_address] Unable to get_contact_by_address address '%s'. Error '%s'", json.encode(address), json.encode(err))
	-- 	return
	-- end
	----The contact address can already exist in another VCC don't use it in that case
	logger.debug("[FullContact::get_contact_by_address] 2")
	local contactsMatching = connection:GetSliceRange(addressId, "ContactsByAddress", "",
		{
			column_names = {},
			slice_range = {
				start = "",
				finish = "",
				reversed = false,
				count = 200
			}
		}
	)
	-- logger.debug("[FullContact::get_contact_by_address] 3")
	-- logger.debug("[FullContact::get_contact_by_address]    success " .. tostring(success))
	-- logger.debug("[FullContact::get_contact_by_address]    addressId " .. tostring(addressId))
	-- logger.debug("[FullContact::get_contact_by_address]    addressType " .. tostring(addressType))
	-- logger.debug("[FullContact::get_contact_by_address]    vcc " .. tostring(vcc))
	-- logger.debug("[FullContact::get_contact_by_address]    address " .. tostring(json.encode(address or {})))
	-- logger.debug("[FullContact::get_contact_by_address]    contactsMatching " .. tostring(contactsMatching))
	-- logger.debug("[FullContact::get_contact_by_address] 4")
	if contactsMatching then
		for column in values(contactsMatching) do
			local contactID = column.name
			--print("contactsMatching", contactID)
			local id, contact_vcc = helpers.SeparateVCC(contactID)
			if contact_vcc == vcc then
				local contact = get_contact(connection, contactID)
				if contact then
					--logger.debug("[FullContact::get_contact_by_address] 5")
					return contact
				end
			end
		end
	end
	--logger.debug("[FullContact::get_contact_by_address] 6")
end

---
-- The CF ContactHandles maps a handle (scoped by type and vcc) to our contact ids.
-- This allows to look up a contact by some value other than addresses or its id.
-- For instance, external ids are a type of handle.
-- Also, webchat's tracking cookie can be used to perform lookups.
--
-- For example, the contact A has an external ID that corresponds to a passport number, which
-- is stored in a client's database. Having this handle mapping allows us to retrieve contact info
-- given one of those ids.
--
-- The relationship is one to one. A contact has at most one type:handle.
--
local function get_contact_by_handle (connection, handle_type, handle, vcc)
	local key = ("%s:%s@%s"):format(handle_type, handle, vcc)
	local contact_info = {}

	--logger.info("[FULLCONTACT:get_contact_by_handle] key '%s'", key)

	-- basic data
	local data = connection:GetSliceRange(key,
		"ContactHandles",
		"",
		{
			column_names = { "contact_id" },
			slice_range = {
				start = "",
				finish = "",
				reversed = false,
				count = 200
			}
		}
	)

	if data then
		pl.tablex.foreach(data, function(v) contact_info[v.name] = v.value end)
	end
	-- if no columns are projected this is a tombstone and should not be returned
	if not contact_info.contact_id then
		return
	end
	local contact = get_contact(connection, contact_info.contact_id)
	if not contact then
		-- it is a logically deleted contact
		return
	end
	return contact
end

---
-- stores the relationship type:handle@vcc -> contact_id
--
local function set_contact_handle (connection, type, handle, vcc, contact_id)
	local key = ("%s:%s@%s"):format(type, handle, vcc)
	--logger.info("[FULLCONTACT:set_contact_handle] - set_contact_handle '%s' -> '%s'", key, contact_id)
	connection:InsertColumn(key, "ContactHandles", "contact_id", contact_id)
end

---
-- deletes the relationship type:handle@vcc -> contact_id
--
local function delete_contact_handle (connection, type, handle, vcc)
	local key = ("%s:%s@%s"):format(type, handle, vcc)
	--logger.info("[FULLCONTACT:delete_contact_handle] - delete_contact_handle '%s'", key)
	connection:RemoveColumn(key, "ContactHandles", "contact_id")
end

---
--
--local function delete_contact (contacts_ks, index_ks, contact_id, options)
local function delete_contact (contacts_ks, contact_id, options)
	options = options or {}

	logger.info("[FULLCONTACT:delete_contact] - Deleting contact: '%s' ", contact_id)

	-- Retrieve the contact, including the logically deleted ones
	local contact = get_contact(contacts_ks, contact_id, { logical = true })
	if not contact then
		return false, "Contact not found"
	end

	-- delete contact addresses
	for address in values(contact.addresses or {}) do
		contacts_ks:RemoveSuperColumn(contact_id, "ContactAddresses", address.AccountId)
		contacts_ks:RemoveColumn(address.AccountId, "ContactsByAddress", contact_id) --update the index the contact does not have this address anymore
	end

	--delete contact campaigns
	for campaign in values(contact.campaigns) do
		--the column value is empty the data is the column name itself
		contacts_ks:RemoveColumn(contact_id, "ContactCampaigns", campaign)
	end
	contacts_ks:RemoveColumn(contact_id, "Contacts")

	if contact.externalId and options.actualContact and options.actualContact.externalId ~= contact.externalId then
		delete_contact_handle(contacts_ks, "external_id", contact.externalId, contact.VCC)
	end

	for handle_type, handle in pairs(contact.handles or {}) do
		if not (options.actualContact and options.actualContact.handles and options.actualContact.handles[handle_type]) then  
			delete_contact_handle(contacts_ks, handle_type, handle, contact.VCC)
		end
	end

	local res = ES.RemoveContact(contact_id, contact.VCC)

	return res, not res and "Unable to delete contact" or ""
end

---
-- Helper to check if a contact other than 'contact' has the same handles.
-- Used to avoid ending up with two contacts with the same external id or handles.
-- When 'obsolete_contact_id' is provided, we allow the conflict, since that contact
-- is going away (the case we are merging contacts)
--
local function check_duplicated_handles (contacts_ks, contact, obsolete_contact_id)
	-- when obsolete_contact_id is just nil, the condition just works

	-- also check if the handles don't belong to another contact (except the one we'll delete)
	if contact.externalId then
		local other_contact = get_contact_by_handle(contacts_ks, "external_id", contact.externalId, contact.VCC)
		if other_contact and (other_contact.Id ~= contact.Id and other_contact.Id ~= obsolete_contact_id) then
			logger.warn("[FULLCONTACT:check_duplicated_handles] - Could not save contact: '%s', external id: '%s' is already in use by contact: '%s'",
				contact.Id, contact.externalId, other_contact.Id)
			return nil, { code = "E_EXT_ID_IN_USE", msg = "External id belongs to another contact",
						  contact_id = other_contact.Id, external_id = contact.externalId }
		end
	end

	for handle_type, handle in pairs(contact.handles) do
		local other_contact = get_contact_by_handle(contacts_ks, handle_type, handle, contact.VCC)
		if other_contact and (other_contact.Id ~= contact.Id and other_contact.Id ~= obsolete_contact_id) then
			logger.warn("[FULLCONTACT:check_duplicated_handles] - Could not save contact: '%s', Handle '%s:%s' is already in use by contact: '%s'",
				contact.Id, handle_type, handle, other_contact.Id)
			return nil, { code = "E_HANDLE_IN_USE", msg = "Handle belongs to another contact",
						  contact_id = other_contact.Id, handle_type = handle_type, handle = handle }
		end
	end

	return true
end

---
-- Helper to check if a contact other than 'contact' has the same addresses.
-- Used to avoid ending up with two contacts with the same address.
-- When 'obsolete_contact_id' is provided, we allow the conflict, since that contact
-- is going away (the case we are merging contacts)
--
local function check_duplicated_addresses (contacts_ks, contact, obsolete_contact_id)
	---- when obsolete_contact_id is just nil, the condition just works
	--LogDebug("[FULLCONTACT:check_duplicated_addresses]     1-> contact.Id: '%s', contact.addresses: '%s'.", contact.Id, json.encode(contact.addresses or {}))
	--LogDebug("[FULLCONTACT:check_duplicated_addresses]     2-> obsolete_contact_id: '%s'.", obsolete_contact_id)
	--local i = 0
	for address in values(contact.addresses or {}) do
		--i = i + 1
		local other_contact = get_contact_by_address(contacts_ks, address.Type, address, contact.VCC)
		--LogDebug("[FULLCONTACT:check_duplicated_addresses]     3.%d-> other_contact.Id: '%s', other_contact.addresses: '%s'.", i, (other_contact or {}).Id, json.encode((other_contact or {}).addresses or {}))
		if other_contact and (other_contact.Id ~= contact.Id and other_contact.Id ~= obsolete_contact_id) then
			logger.warn("[FULLCONTACT:check_duplicated_addresses] - Could not save contact: '%s', address: %s is already in use by contact: '%s'", contact.Id, address.displayable, other_contact.Id)
			return nil, {
				code = "E_ADDR_IN_USE",
				msg = "Address belongs to another contact",
				contact_id = other_contact.Id,
				address = address
			}
		end
	end
	return true
end

--[[

]]--
local function ValidateAddresses (contact, currentContact, options)
	options = options or {}
	local invalidAddresses = {}
	local successAddresses = {}
	local mapCurrentContactAddresses = {}

	--logger.debug("[FULLCONTACT:ValidateAddresses]     --------------> currentContact: '%s'.", helpers.JSON_Encode(currentContact))

	if currentContact then
		-- Si estoy actualizando el contacto, me creo un map de las address que ya tiene configuradas.
		--   Con este map luego valido si el mismo address posteado, es invalido, y si es asi, tengo que agregarlo a la lista de succesAddresses, para que la misma no se borre de la BD.
		for _, address in ipairs(currentContact.addresses or {}) do
			if address.AccountId then
				mapCurrentContactAddresses[address.AccountId:lower()] = address
			end
		end
		--logger.debug("[FULLCONTACT:ValidateAddresses] -------> mapCurrentContactAddresses: '%s'", helpers.JSON_Encode(mapCurrentContactAddresses))
	end

	--logger.debug("[FULLCONTACT:ValidateAddresses] (START) ---> Validate contact addresses: '%s'", helpers.JSON_Encode(contact))
	for address in values(contact.addresses or {}) do
		--logger.debug("[FULLCONTACT:ValidateAddresses]  ---> Validate ContactAddresses: '%s'", helpers.JSON_Encode(address))
		local address_type = utf8.lower(address.Type)
		local success = true
		local resp, isGlobalNumber
		local normalizedAddressInfo = helpers.BuildAddressInstance(address.Type, address, contact.VCC)
		address.AccountId = normalizedAddressInfo.AccountId

		if address_type == "phone" then
			--LogDebug("-----------> normalizedAddressInfo: %s", json.encode(normalizedAddressInfo or {}))
			if (normalizedAddressInfo.Validation or {}).Success then
				--LogDebug("[FULLCONTACT:ValidateAddresses] --> BuildAddressInstance-> helpers.validation:IsPhone --> OK:  number: %s, country: %s, area: %s, resp: %s", address.Number or address.address, address.Country, address.Area, json.encode(normalizedAddressInfo.Validation))
				address.Number       = normalizedAddressInfo.Number       -- dialGlobal  (FROM: helpers.BuildAddressInstance -> AddressMapper.phone -> helpers.validation.IsPhone)
				address.LocalNumber  = normalizedAddressInfo.LocalNumber  -- dialLocal   (FROM: helpers.BuildAddressInstance -> AddressMapper.phone -> helpers.validation.IsPhone)
				address.SimpleNumber = normalizedAddressInfo.SimpleNumber -- number      (FROM: helpers.BuildAddressInstance -> AddressMapper.phone -> helpers.validation.IsPhone)
				address.displayable  = normalizedAddressInfo.displayable  -- displayable (FROM: helpers.BuildAddressInstance -> AddressMapper.phone -> helpers.validation.IsPhone)
			--else
				--LogWarning("[FULLCONTACT:ValidateAddresses] --> BuildAddressInstance-> helpers.validation:IsPhone --> INVALID:  number: %s, country: %s, area: %s, resp: %s", address.Number or address.address, address.Country, address.Area, json.encode(normalizedAddressInfo.Validation))
			end
		end
		if success then
			--LogDebug("[FULLCONTACT:ValidateAddresses]   1--> helpers.validation:IsPhone --> Add success address '%s' en SUCCESS list.", address.AccountId or "")
			table.insert(successAddresses, address)
		else
			local invalidAddress = address
			table.insert(invalidAddresses, {
				address = invalidAddress,
				err = resp
			})
			local accountId = (invalidAddress.AccountId or ""):lower()
			--LogDebug("[FULLCONTACT:ValidateAddresses]   2--> helpers.validation:IsPhone --> Add invalid address '%s' en INVALID list.", accountId)
			-- 
			local mustAddInSuccessList = false
			mustAddInSuccessList = mapCurrentContactAddresses[accountId]
			if mustAddInSuccessList then
				--LogDebug("[FULLCONTACT:ValidateAddresses]   3--> helpers.validation:IsPhone --> Add CURRENT invalid address '%s' en INVALID list.", accountId)
				table.insert(successAddresses, invalidAddress)
			end
		end
	end
	-- OK - actualizo las addresses
	--logger.debug("[FULLCONTACT:ValidateAddresses] (END) ---> successAddresses '%s', invalidAddresses '%s'", helpers.JSON_Encode(successAddresses), helpers.JSON_Encode(invalidAddresses))
	if options.mustReplaceWithValidatedAddresses then
		contact.addresses = successAddresses
	end
	return true, successAddresses, invalidAddresses
end

---
-- @return ok, a table with addresses to be indexed and the current contact if it already existed
-- @return nil plus an error
--
local function save_contact (contacts_ks, contact, options)
	local id = contact.Id
	-- logger.info("[FULLCONTACT:save_contact] -> (START) Saving contactId: '%s'", id)
	-- logger.debug("[FULLCONTACT:save_contact]   ->  Saving contact: '%s'", helpers.JSON_Encode(contact))

	options = options or {}
	if options.check_duplicated_addresses == nil then
		options.check_duplicated_addresses = true
	end
	if options.check_duplicated_handles == nil then
		options.check_duplicated_handles = true
	end
	if options.mustReplaceWithValidatedAddresses ~= true then
		options.mustReplaceWithValidatedAddresses = false
	end

	--logger.debug("[FULLCONTACT:save_contact]   ->  options: '%s'",  helpers.JSON_Encode(options))

	--logger.debug("[FULLCONTACT:save_contact]     1-> postedContact.addresses: '%s'.", helpers.JSON_Encode(contact.addresses))

	-- Recupero contacto en caso que exista
	local currentContact = get_contact(contacts_ks, id)
	--logger.debug("[FULLCONTACT:save_contact]     1.1-> currentContact: '%s'.", helpers.JSON_Encode(currentContact))

	------------------------------------------------
	-- valido & sanitizo las addresses
	local success, successAddresses, invalidAddresses = ValidateAddresses(contact, currentContact, options)
	local errInvalidAddresses
	if #invalidAddresses > 0 then
		errInvalidAddresses = {
			code = "E_THERE_ARE_INVALID_ADDRESSES",
			msg = "There are invalid addresses",
			contact_id = contact.Id,
			invalidAddresses = invalidAddresses
		}
		logger.warn("[FULLCONTACT:save_contact] Warning saving contact: '%s'. There are %d invalid addresses. Err: %s", contact.Id, #errInvalidAddresses.invalidAddresses, helpers.JSON_Encode(errInvalidAddresses))
	end

	-- check to see if the addresses don't belong to another contact
	-- TODO: this should be outside this method. it hurts composability
	-- See the shitty flag I had to add...
	if options.check_duplicated_addresses then
		local ok, err = check_duplicated_addresses(contacts_ks, contact)
		if not ok then
			logger.warn("[FULLCONTACT:save_contact] Unable to Save contact: '%s'. Duplication addresses. Err: %s", contact.Id, helpers.JSON_Encode(err))
			return false, err
		end
	end

	if options.check_duplicated_handles then
		local ok, err = check_duplicated_handles(contacts_ks, contact)
		if not ok then
			logger.warn("[FULLCONTACT:save_contact] Unable to Save contact: '%s'. Duplication handles. Err: %s", contact.Id, helpers.JSON_Encode(err))
			return false, err
		end
	end

	--
	-- Now, we'll store in Cassandra the data of the contact.
	--
	local values_to_store = get_values_to_store(contact)

	-- remove values that have been deleted (present in old contact, not in new one)
	--logger.debug("[FULLCONTACT:save_contact]    -> (Borro) Recorro las columnas del contacto existente y remuevo las que el contacto nuevo no incluye.")
	pl.tablex.foreach(get_values_to_store(currentContact or {}), function(value, key)
		if contact[key] == nil then
			--logger.debug("[FULLCONTACT:save_contact]        -> borro la columna: '%s'", key)
			contacts_ks:RemoveColumn(id, "Contacts", key)
		end
	end)

	-- write values to Cassandra if there is no existing contact or if the value changed
	--logger.debug("[FULLCONTACT:save_contact]    -> (Alamaceno) Recorro las columnas a almacenar del nuevo contacto y si no existen en el viejo contacto o si son diferentes las agrego.")
	pl.tablex.foreach(values_to_store, function(value, key)
		if not currentContact or value ~= currentContact[key] then
			--logger.debug("[FULLCONTACT:save_contact]        -> agrego la columna: '%s' = '%s'", key, tostring(value))
			contacts_ks:InsertColumn(id, "Contacts", key, tostring(value))
		end
	end)

	-- deal with handles explicitly. Encode those in json (when table is not empty)
	if type(contact.handles) == "table" and next(contact.handles) then
		contacts_ks:InsertColumn(id, "Contacts", "__handles", helpers.JSON_Encode(contact.handles))
	end

	-- Saving the contact deletes the avatar in order to trigger re-searching
	contacts_ks:RemoveColumn(id, "Contacts", "Avatar")

	local new_accounts = {}
	for address in values(contact.addresses or {}) do
		local address_type = utf8.lower(address.Type)
		local data_to_store = AddressMapper[address_type](address, contact.VCC)
		if data_to_store then
			local _, address_id = helpers.BuildAddressId(address_type, address, contact.VCC)
			pl.tablex.foreach(data_to_store, function(value, key)
				value = type(value) == "table" and helpers.JSON_Encode(value) or value
				--logger.debug("[FULLCONTACT:save_contact]     ---> Update 'ContactAddresses', address_id: '%s', key: '%s', value: '%s' ", address_id, key, value)
				contacts_ks:InsertSuperColumn(id, "ContactAddresses", address_id, key, value)
			end)
			-- update the index of contacts by address
			new_accounts[address_id] = true
			contacts_ks:InsertColumn(address_id, "ContactsByAddress", id, "")
		--else
		--	logger.debug("[FULLCONTACT:save_contact]     ---> Unable to include address '%s'.", json.encode(address))
		end
	end

	--logger.debug("[FULLCONTACT:save_contact]     2-> new_accounts: '%s'.", helpers.JSON_Encode(new_accounts))

	-- store the new external id when present
	if contact.externalId then
		set_contact_handle(contacts_ks, "external_id", contact.externalId, assert(contact.VCC), id)
	end

	for handle_type, handle in pairs(contact.handles) do
		set_contact_handle(contacts_ks, handle_type, handle, assert(contact.VCC), id)
	end

	-- save contact campaigns
	local new_campaigns = {}
	for campaign in values(contact.campaigns or {}) do
		-- the column value is empty, the data is the column name itself
		contacts_ks:InsertColumn(id, "ContactCampaigns", campaign, "")
		new_campaigns[campaign] = true
	end

	-- delete missing parts of the contact, campaign and addresses
	if currentContact then
		-- delete the old external id if any
		if currentContact.externalId and currentContact.externalId ~= contact.externalId then
			delete_contact_handle(contacts_ks, "external_id", currentContact.externalId, currentContact.VCC)
		end

		-- delete the old handles, if any
		for handle_type, handle in pairs(currentContact.handles or {}) do
			if not contact.handles[handle_type] then
				delete_contact_handle(contacts_ks, handle_type, handle, currentContact.VCC)
			end
		end

		-- Si trae un Address entonces borro lo viejo. Si viene vacio entonces mantengo las address anteriores.
		--local contactAddresses = contact.addresses or {}
		--logger.debug("[FULLCONTACT:save_contact]     3-> currentContact.addresses: '%s'.", helpers.JSON_Encode(currentContact.addresses))
		--if not helpers.TableIsEmpty(contactAddresses) then
			for address in values(currentContact.addresses or {}) do
				if not new_accounts[address.AccountId] then
					--logger.debug("[FULLCONTACT:save_contact]       ---> Remove related CFs 'ContactAddresses' & 'ContactsByAddress'. contactId: '%s', accountId: '%s' ", id, address.AccountId)
					contacts_ks:RemoveSuperColumn(id, "ContactAddresses", address.AccountId)
					contacts_ks:RemoveColumn(address.AccountId, "ContactsByAddress", id) --update the index the contact does not have this address anymore
				end
			end
		--end
		for campaign in values(currentContact.campaigns or {}) do
			if not new_campaigns[campaign] then
				contacts_ks:RemoveColumn(id, "ContactCampaigns", campaign)
			end
		end
	end

	--logger.info("[FULLCONTACT:save_contact] - (END) Success Saved contact: '%s'", id)
	local newContact = get_contact(contacts_ks, id)
	--logger.debug("[FULLCONTACT:save_contact]     1.2-> newContact: '%s'.", helpers.JSON_Encode(newContact))

	if currentContact then
		currentContact = get_contact(contacts_ks,id)
	end
	return true, currentContact, errInvalidAddresses
end

---
--
local function add_contact_search_engine (contact, elasticType, options)
	--logger.info("[FULLCONTACT:add_contact_search_engine] - elasticType: '%s', contact: '%s'", elasticType, helpers.JSON_Encode(contact))
	options = options or {}

	contact.addresses = build_address_to_index(contact.addresses, contact.VCC)
	local success, contactId, result
	if elasticType == "CONTACTS_DIALER" then
		success, contactId, result = ES.Dialer:StoreContact(contact, options)
	else
		success, contactId, result = ES.StoreContact(contact, elasticType)
	end
	if not success then
		-- error
		local err = result
		return false, contactId, err
	end
	-- ok
	local contactAlreadyExist = result
	return true, contactId, contactAlreadyExist
end

---
--
--local function merge_contacts (contacts_ks, index_ks, contact, obsolete_contact_id)
local function merge_contacts (contacts_ks, contact, obsolete_contact_id)
	local id = contact.Id

	logger.info("[FULLCONTACT:merge_contacts] ---> Merging contacts: '%s', obsolete: '%s'", contact.Id, obsolete_contact_id)

	-- check to see if the addresses don't belong to another contact (except the one we'll delete)
	local ok, err = check_duplicated_addresses(contacts_ks, contact, obsolete_contact_id)
	if not ok then return ok, err end

	-- also check if the handles don't belong to another contact (except the one we'll delete)
	local ok, err = check_duplicated_handles(contacts_ks, contact, obsolete_contact_id)
	if not ok then return ok, err end

	local ok, currentContact = save_contact(contacts_ks, contact, { check_duplicated_addresses = false, check_duplicated_handles = false })
	if ok then
		add_contact_search_engine(contact)

		if delete_contact(contacts_ks, obsolete_contact_id, {actualContact = contact}) then
			--logger.info("[FULLCONTACT:merge_contacts] Success merging contact: '%s' was saved and contact: '%s' became obsolete", contact.Id, obsolete_contact_id)
			return true
		else
			logger.warn("[FULLCONTACT:merge_contacts] Contacts Merge - Unable to delete the obsolete contact %s", obsolete_contact_id)
			return nil, {
				code = "E_CANNOT_DELETE_OBSOLETE",
				msg = "Unable to delete the obsolete contact",
				obsolete_contact_id = obsolete_contact_id
			}
		end
	else
		logger.warn("[FULLCONTACT:merge_contacts] - Contacts Merge - Unable to save newly merged contact '%s'", id)
		local err = currentContact
		if type(err) == "table" then
			return nil, err
		end
		return nil, {
			code = "E_FAIL",
			msg = "Unable to save newly merged contact",
			contact_id = id,
			info = err
		}
	end
end

---
-- Saves an avatar for the contact
--
function public.SaveAvatar (contact_id, avatar)
	return with_connection_do(function(connection)
		local contact, err = get_contact(connection, contact_id)
		if contact then
			connection:InsertColumnWithSize(contact.Id, "Contacts", "Avatar", avatar.image, avatar.size)
			return true
		end
		return nil, err
	end)
end

---
-- Change Contact State
--
function public.ChangeContactState (contact_id, state)
	with_connection_do(function(connection)
		connection:InsertColumn(contact_id, "Contacts", "state", state)
	end)
end

---
--
function public.SetDatasource (datasource)
	m_datasource.cassandra = datasource.cassandra
	m_datasource.redis = datasource.redis
	m_datasource.cassandraInfo = datasource.cassandraInfo
	m_datasource.timeServer = datasource.timeServer
	m_datasource.cluster = datasource.cluster
	helpers.SetupConnections(m_datasource)
	ES.SetConnection(datasource)
end

---
--
function public.GetDatasource ()
	return m_datasource
end

---
-- Retrieves a Contact given its id.
--
-- @tparam table options A table containing options
-- @tparam boolean options.logical Retrieve the contact even if it is logically deleted.
--
function public.GetContact (contact_id, options)
	--logger.debug("[FULLCONTACT:public.GetContact] - contact_id '%s'", contact_id)
	return with_connection_do(get_contact, contact_id, options)
end

function public.GetContactByAddress (addressType, address, vcc, options)
	return with_connection_do(get_contact_by_address, addressType, address, vcc, options)
end

function public.GetContactByExternalId (external_id, vcc, options)
	return with_connection_do(get_contact_by_handle, "external_id", external_id, vcc, options)
end

function public.GetContactByHandle (handle_type, handle, vcc, options)
	return with_connection_do(get_contact_by_handle, handle_type, handle, vcc, options)
end

---
-- saves a contact and adds it to the ELASTIC SEARCH for further searching
--
function public.Save (contact, options)
	options = options or {}
	--logger.info("[FULLCONTACT:Save] 1- (ENTRA) Saving contact '%s', options: '%s.", helpers.JSON_Encode(contact), helpers.JSON_Encode(options))
	local ok, currentContact, partialError = with_connection(save_contact)(contact, options)
	--logger.info("[FULLCONTACT:Save] 1-2-   ok: '%s', currentContact: %s.", ok, helpers.JSON_Encode(currentContact))
	if not ok then
		local err = currentContact
		return false, err
	end
	local success = false
	local newContact = currentContact or contact
	if newContact then
		success = add_contact_search_engine(newContact)
		setmetatable(newContact, Contact_mt)
		if newContact.addresses then
			setmetatable(newContact.addresses, Addresses_mt)
		end
	end
	--logger.info("[FULLCONTACT:Save] 3- (SALE) Saved contact success: '%s', newContact: %s, partialError: %s.", success, helpers.JSON_Encode(newContact), helpers.JSON_Encode(partialError))
	return success, newContact, partialError
end

---
-- Deletes contact.
--
-- @tparam table options A table containing options
-- @tparam boolean options.logical Deletes the contact 'logically'. i.e. it is not actually deleted, but it
-- won't show up in searches or gets.
--
function public.DeleteContact (id, options)
	return with_connection(delete_contact)(id, options)
end

function public.Search (params)
	--logger.info("[FULLCONTACT:Search] - Searching contacts in VCC '%s' using ELASTIC SEARCH: campaigns '%s', filter '%s', usersFavorite '%s' start '%s', rows per page '%s', sorted: '%s'.", params.vcc, helpers.JSON_Encode(params.campaigns), helpers.JSON_Encode(params.filter), helpers.JSON_Encode(params.usersFavorite or {}), params.startRow, params.rowsPerPage, (params.sort and helpers.JSON_Encode(params.sort) or "NONE"))

	local query = {
		filter = params.filter,
		campaigns = params.campaigns,
		startRow = params.startRow,
		rowsPerPage = params.rowsPerPage,
		vcc = params.vcc,
		sort = params.sort,
		usersFavorite = params.usersFavorite
	}

	local success, response = ES.SearchContacts(query)
	if not success then
		local err = response
		return false, err
	end

	return true, response
end


function public.AddToFavorites (userId, contactId)
	--logger.info("[FULLCONTACT:AddToFavorites] User %s add favorite to contact %s.", userId, contactId)

	local simpleUserId, vcc = helpers.SeparateVCC(userId)
	if not vcc then
		_, vcc = helpers.SeparateVCC(contactId)
	end
	assert(vcc)

	local success, contact = public.GetContactFromES(contactId, vcc)
	if not success then
		logger.error("[FULLCONTACT:AddToFavorites] Unable to add contact favorite. Contact '%s' does not exist", contactId)
		return false
	end

	simpleUserId = utf8.lower(simpleUserId)
	usersFavorite = contact.UsersFavorite or {}
	local alreadyExist = false
	for _, usersFavoriteId in ipairs(usersFavorite) do
		if usersFavoriteId == simpleUserId then
			alreadyExist = true
			break
		end
	end
	if alreadyExist then
		logger.warn("[FULLCONTACT:AddToFavorites] Unable to add contact '%s' favorite for user'%s'. Already exist.", contactId, userId)
		return false
	end

	table.insert(usersFavorite, utf8.lower(simpleUserId))
	contact.UsersFavorite = usersFavorite

	success = public.AddContactToES(contact)
	if not success then
		logger.error("[FULLCONTACT:AddToFavorites] Unable to add contact '%s' favorite. ELASTIC SEARCH error", contactId)
		return false
	end

	--logger.info("[FULLCONTACT:AddToFavorites] Success Added Users '%s' to Favorite Contact %s.", userId, contactId)

	return true
end

function public.RemoveFromFavorites (userId, contactId)
	--logger.info("[FULLCONTACT:RemoveFromFavorites] - User %s remove favorite to contact %s.", userId, contactId)

	local simpleUserId, vcc = helpers.SeparateVCC(userId)
	if not vcc then
		_, vcc = helpers.SeparateVCC(contactId)
	end
	assert(vcc)

	local success, contact = public.GetContactFromES(contactId, vcc)
	if not success then
		logger.error("[FULLCONTACT:RemoveFromFavorites] Unable to remove contact favorite. Contact '%s' does not exist", contactId)
		return false
	end

	--logger.info("[FULLCONTACT:RemoveFromFavorites] Current users favorites for contactId: '%s' contact.UsersFavorite '%s' ", contactId, helpers.JSON_Encode(contact.UsersFavorite or {}))

	local indexToRemove
	local simpleUserId = utf8.lower(simpleUserId)
	for i, userFavoriteId in ipairs(contact.UsersFavorite) do
		if userFavoriteId == simpleUserId then
			indexToRemove = i
			break
		end
	end
	if not indexToRemove then
		logger.warn("[FULLCONTACT:RemoveFromFavorites] Unable to remove contact '%s' favorite for user'%s'. Contact is not favorite for User.", contactId, userId)
		return false
	end

	table.remove(contact.UsersFavorite, indexToRemove)

	success = public.AddContactToES(contact)
	if not success then
		logger.error("[FULLCONTACT:RemoveFromFavorites] Unable to remove contact '%s' favorite. ELASTIC SEARCH error", contactId)
		return false
	end

	--logger.info("[FULLCONTACT:RemoveFromFavorites] Success Removed Users '%s' to Favorite Contact %s.", userId, contactId)

	return true
end

--[[
	Obtiene los favorites contacts, paginados desde ELASTIC SEARCH.
]]--
function public.GetUserFavorites (params)
	local fullUserId = utf8.lower(helpers.AddVCC(params.user, params.vcc))
	--logger.info("[FULLCONTACT:GetUserFavorites] Retrieving Favorite Contacts for User: '%s'.", fullUserId)

	local usersFavorite = {}
	table.insert(usersFavorite, fullUserId)

	local query = {
		usersFavorite = usersFavorite,
		vcc = params.vcc,
		startRow = params.startRow,
		rowsPerPage = params.rowsPerPage,
		sort = {{Name = {order = "asc"}}}
	}
	local success, favorites = public.Search(query)
	if not success then
		logger.warn("[FULLCONTACT:GetUserFavorites] Unable to retrieve Favorite Contacts for User: '%s'.", fullUserId)
		return
	end
	--logger.info("[FULLCONTACT:GetUserFavorites] Retrieved Favorite '%s' Contacts for User: '%s'.", #favorites.contacts, fullUserId)
	return favorites
end

---
-- Merge involves saving contact and then deleting the obsolete contact
--
function public.MergeContacts (contact, obsolete_contact_id)
	return with_connection(merge_contacts)(contact, obsolete_contact_id)
end


---
-- Changes the state of the contact (IM mostly)
--
function public.ChangeState (contact_id, new_state)
	return with_connection_do(function(connection)
		return connection:InsertColumn(contact_id, "Contacts", "state", new_state)
	end)
end

--[[
********************************************************************************************************
Funciones (AddContactToES, RemoveContactFromES) que quedan expuestas en el API,
   para que puedan ser utilizar desde los migrations (Ej: "20180524T235959_7726_AddContactsSearchEngineTOElasticSearch.sh")
   mediante el API de FULLCONTACT (fullcontact.lua)
********************************************************************************************************
]]--

--------------------------
-- Funcion que crea el Indice en el Motor de búsqueda Elastic Search
--------------------------
function public.CreateESIndex (vcc, elasticType)
	return ES.CreateIndex(vcc, elasticType)
end

--------------------------
-- Funcion que elimina el Indice del Motor de búsqueda Elastic Search
--------------------------
function public.DropESIndex (vcc, elasticType)
	return ES.DropIndex(vcc, elasticType)
end

--------------------------
-- Funcion que crea el Template en el Motor de búsqueda Elastic Search
--------------------------
function public.CreateESTemplate (elasticType)
	return ES.CreateTemplate(elasticType)
end

--------------------------
-- Funcion que retorna el nombre del Indice utilizado en el Motor de búsqueda Elastic Search
--------------------------
function public.GetESIndexName (vcc, elasticType)
	return ES.GetIndexName(vcc, elasticType)
end

--------------------------
-- Agrega un contacto al Motor de búsqueda Elastic Search
--------------------------
function public.AddContactToES (contact, elasticType)
	return add_contact_search_engine(contact, elasticType)
end

--------------------------
-- Remueve un contacto del Motor de búsqueda Elastic Search
--------------------------
function public.RemoveContactFromES (contactId, vcc)
	contactId = helpers.AddVCC(contactId, vcc)
	return ES.RemoveContact(contactId, vcc)
end

--------------------------
-- Obtiene un contacto desde Motor de búsqueda Elastic Search
--------------------------
function public.GetContactFromES (contactId, vcc)
	assert(contactId)
	local simpleId, contactVCC = helpers.SeparateVCC(contactId)
	vcc = vcc or contactVCC
	local fullId = helpers.AddVCC(simpleId, vcc)
	return ES.GetContact(fullId, vcc)
end

--------------------------
-- Lista todos los contactos existentes en Cassandra
--------------------------
local m_allVCCsContacts
function public.ListAllContactsFromCassandra (vcc, options)
	options = options or {}

	--logger.info("[FULLCONTACT:ListAllContactsFromCassandra] Wait moment please ... Loading All Contacts from Cassandra for VCC '%s'.", vcc)

	if not m_allVCCsContacts then
		local keyspace = m_datasource.cassandra:CassandraConnection("Contacts")
		m_allVCCsContacts = helpers.GetAllColumns({
			keyspace = keyspace,
			columnFamily = "Contacts",
			startKey = "",
			endKey = "",
			rows_per_chunk = options.rows_per_chunk or 1000,
			sleep_between_chunk_sec = options.sleep_between_chunk_sec or nil
		}) or {}
		keyspace:Close()
	end

	vcc = vcc and utf8.lower(vcc) or nil
	local contacts = {}
	local chunkTrace = 5000
	local i = 1
	for idAtVCC, contact in pairs(m_allVCCsContacts) do
		local id, contactVCC = helpers.SeparateVCC(idAtVCC)
		contactVCC = utf8.lower(contactVCC or "")
		if not vcc or vcc == contactVCC then
			i = i + 1
			if i == chunkTrace then
				chunkTrace = chunkTrace + 5000
				--logger.info("[FULLCONTACT:ListAllContactsFromCassandra] Loaded %s Contacts of many for VCC '%s'... wait moment.", i, vcc)
			end
			if options.allInfo then
				local contactWithAllInfo = public.GetContact(idAtVCC, {logical = true})
				contacts[idAtVCC] = contactWithAllInfo
			else
				contacts[idAtVCC] = contact
			end
		end
	end

	--logger.info("[FULLCONTACT:ListAllContactsFromCassandra] Finished Loaded %s Contacts from Cassandra for VCC '%s'.", i, vcc)

	return contacts
end

--------------------------
-- Lista un grupo de contactos existentes en Cassandra
--------------------------
function public.ListContactsChunkFromCassandra (options, cassandraConnection)
	options = options or {}

	local closeConnection = false
	local startKeyParam = options.startKey or ""
	local keyspace
	local contacts = {}
	local i = 0
	local lastId = ""
	local totalRow = options.rowCount or 1000
	
	--logger.info("[FULLCONTACT:ListContactsChunkFromCassandra] Wait moment please ... Loading %d Contacts from Cassandra. Start on Key: %s",totalRow, startKeyParam)
	
	if not cassandraConnection then
		keyspace = m_datasource.cassandra:CassandraConnection("Contacts")
		closeConnection = true
	else
		keyspace = cassandraConnection
	end
	
	local m_allVCCsContacts = keyspace:GetRangeSlices("Contacts",  --column family
		"",  --supercolumn
		startKeyParam, --startkey
		"", --endkey
		{ --predicate
			column_names = {}, 
			slice_range = { 
						start = "", 
						finish = "", 
						reversed = false, 
						count = 1000}
		},
		totalRow --total keys
	)
	
	if m_allVCCsContacts then
		local chunkTrace = 5000
		for idAtVCC, contact in m_allVCCsContacts:Iterator() do
			i = i + 1
			if i == chunkTrace then
				chunkTrace = chunkTrace +  5000
				--logger.info("[FULLCONTACT:ListContactsChunkFromCassandra] Loaded %s Contacts of many ... wait moment.", i)
			end
			if options.allInfo then
				local contactWithAllInfo = public.GetContact(idAtVCC, {logical = true})
				contacts[idAtVCC] = contactWithAllInfo
			else
				contacts[idAtVCC] = contact
			end
			lastId = idAtVCC
		end
	end
	if closeConnection then
		keyspace:Close()
	end
	--logger.info("[FULLCONTACT:ListContactsChunkFromCassandra] Finished Loaded %s Contacts from Cassandra.", i)

	return contacts, lastId
end

-------------------------
-- Address Book
--------------------------
public.AddressesBook = {}
-------------------

local function MakeChannelId (AddressIdWithVcc, channel)
	local id, vcc = helpers.SeparateVCC(AddressIdWithVcc)
	assert(vcc)
	local channelWithVcc = helpers.AddVCC(channel, vcc):upper()
	return channelWithVcc
end

local function AddAddressesBookRelations (keyspace, params)
	assert(params)
	assert(params.Channel)
	assert(params.AddressId)
	local channelWithVcc = MakeChannelId(params.AddressId, params.Channel)
	--logger.info("[FULLCONTACT:AddAddressesBookRelations] - Adding address relations '%s'", helpers.JSON_Encode(params))
	keyspace:InsertColumn(channelWithVcc, "AddressesBookByChannel", params.AddressId, "")
end

local function RemoveAddressesBookRelations (keyspace, params)
	assert(params)
	assert(params.Channel)
	assert(params.AddressId)
	local channelWithVcc = MakeChannelId(params.AddressId, params.Channel)
	--logger.info("[FULLCONTACT:RemoveAddressesBookRelations] - Removing address relations '%s'", helpers.JSON_Encode(params))
	keyspace:RemoveColumn(channelWithVcc, "AddressesBookByChannel", params.AddressId)
end

---
--
public.AddressesBook.List = function(options)
	options = options or {}

	--logger.info("[FULLCONTACT.AddressesBook:List] Wait moment please ... Listing Addresses Book from Cassandra for '%s'.", helpers.JSON_Encode(options))

	local keyspace = m_datasource.cassandra:CassandraConnection("Contacts")
	local allAddressesBook = helpers.GetAllColumns({
		keyspace = keyspace,
		columnFamily = "AddressesBook",
		startKey = "",
		endKey = "",
		rows_per_chunk = options.rows_per_chunk or 1000,
		sleep_between_chunk_sec = options.sleep_between_chunk_sec or nil
	}) or {}
	keyspace:Close()

	local vcc = options.vcc and utf8.lower(options.vcc) or nil
	local addressesBook = {}
	local chunkTrace = 500
	local i = 1

	for idAtVCC, addressBook in pairs(allAddressesBook) do
		local id, addressVCC = helpers.SeparateVCC(idAtVCC)
		addressVCC = utf8.lower(addressVCC or "")
		--logger.debug("[FULLCONTACT.AddressesBook:List] addressVCC %s  vcc '%s'... wait moment.", addressVCC, vcc)
		if vcc == addressVCC then
			i = i + 1
			if i == chunkTrace then
				chunkTrace = chunkTrace + 500
				--logger.debug("[FULLCONTACT.AddressesBook:List] Loaded %s Address Book of many for VCC '%s'... wait moment.", i, vcc)
			end
			if options.withContactsInfo then
				local addressType = addressBook.Type
				--local success, addressId = helpers.BuildAddressId(addressType, addressBook, vcc)
				local contact = public.GetContactByAddress(addressType, addressBook, addressVCC, {logical = true})
				addressBook.Contacts = contact or {}
			end
			table.insert(addressesBook, addressBook)
		end
	end

	--logger.info("[FULLCONTACT.AddressesBook:List] Finished List. Total %s Addresses Book from Cassandra for '%s'.", i, helpers.JSON_Encode(options))

	return true, addressesBook
end

---
--
public.AddressesBook.ListByChannel = function(options)
	options = options or {}

	--logger.info("[FULLCONTACT.AddressesBook:ListByChannel] Wait moment please ... Listing Addresses Book By Channel from Cassandra for '%s'.", helpers.JSON_Encode(options))

	assert(options.vcc)
	assert(options.channel)
	local channelWithVcc = helpers.AddVCC(options.channel, options.vcc):upper()

	local keyspace = m_datasource.cassandra:CassandraConnection("Contacts")
	local addressesBookByChannel = helpers.GetAllColumns({
		keyspace = keyspace,
		columnFamily = "AddressesBookByChannel",
		key = channelWithVcc,
		--rowCount = 10000,
		rows_per_chunk = options.rows_per_chunk or 1000,
		sleep_between_chunk_sec = options.sleep_between_chunk_sec or nil
	}) or {}

	--logger.info("[FULLCONTACT.AddressesBook:ListByChannel] addrechannelWithVccssVCC %s  vcc '%s'.", channelWithVcc, helpers.JSON_Encode(addressesBookByChannel))

	local addressesBook = {}
	local i = 0
	for addressIdWithVcc in pairs(addressesBookByChannel) do
		i = i + 1
		-- Obtengo el current Address
		local success, currentAddressesBook = public.AddressesBook.Get(addressIdWithVcc, options.withContactsInfo, keyspace)
		if success then
			table.insert(addressesBook, currentAddressesBook)
		end
	end
	keyspace:Close()

	--logger.info("[FULLCONTACT.AddressesBook:ListByChannel] Finished List. Total %s Addresses Book By Channel from Cassandra for '%s'.", i, helpers.JSON_Encode(options))
	return true, addressesBook
end

---
--
public.AddressesBook.Get = function(addressId, withContactsInfo, keyspace)
	assert(addressId)

	addressId = addressId:upper()
	--logger.info("[FULLCONTACT.AddressesBook:Get] Wait moment please ... Retrieving Addresses Book Id '%s'.", addressId)

	local addressesBook = helpers.GetAllColumns {
		keyspace = keyspace or "Contacts",
		columnFamily = "AddressesBook",
		key = addressId,
		rowCount = 10000
	}
	--logger.info("[FULLCONTACT.AddressesBook:Get] Retrieved Addresses Book from Cassandra '%s'.", helpers.JSON_Encode(addressesBook or {}))

	if helpers.TableIsEmpty(addressesBook) then
		logger.error("[FULLCONTACT:AddressesBook:Get] Addresses Book '%s' does not exist", addressId)
		return false
	end

	if withContactsInfo then
		local addressType = addressesBook.Type
		local addressVCC = addressesBook.VCC
		local contact = public.GetContactByAddress(addressType, addressesBook, addressVCC, {logical = true})
		addressesBook.Contacts = contact or {}
	end

	return true, addressesBook
end

---
--
public.AddressesBook.Save = function(addressBook)
	assert(addressBook)
	assert(addressBook.VCC)
	assert(addressBook.Type)

	local vcc = addressBook.VCC
	local success, addressId, err = helpers.BuildAddressId(addressBook.Type, addressBook, vcc)
	if not success then
		logger.warn("[FullContact::AddressesBook:Save] Unable to Save address book '%s'. Error '%s'", json.encode(addressBook), json.encode(err))
		return false, "INVALID_PHONE_NUMBER", err
	end

	addressBook.AddressId = addressId
	local addressIdWithVcc = helpers.AddVCC(addressId, vcc):upper()

	logger.info("[FULLCONTACT.AddressesBook:Save] Wait moment please ... Saving Addresses Book Id '%s' in Cassandra. Data: '%s'.", addressIdWithVcc, helpers.JSON_Encode(addressBook))

	local function Save (keyspace)
		if addressBook.OldAddressInfo then
		-- ACA ingresa solo cuando modificamos !!
		--
		-- En el atributo "OldAddressInfo" viene la informacion desde la web, con la información de la entidad antes de ser modificada.
		--  Esto ya que desde la web, se esta permitiendo cambiar los campos claves del "Address" (no debería poder cambiarse pero bueno).
		--  Ej: se cambia la dirección de correo.
		--
			local success, oldAddressId, err = helpers.BuildAddressId(addressBook.Type, addressBook.OldAddressInfo, vcc)
			local oldAddressIdWithVcc = helpers.AddVCC(oldAddressId, vcc):upper()
			if not success or oldAddressIdWithVcc ~= addressIdWithVcc then
				local newAddressIdWithVcc = addressIdWithVcc
				local alreadyExist, alreadyAddressInfo = public.AddressesBook.Get(newAddressIdWithVcc)
				if alreadyExist then
					--logger.info("[FULLCONTACT.AddressesBook:Save] Unable to Modify Address Info. Address Key '%s' Already exist.: Already Address Info: '%s'.", newAddressIdWithVcc, helpers.JSON_Encode(alreadyAddressInfo))
					return false, "ALREADY_EXIST_ADDRESSES_BOOK"
				end
				-- Si el addres Id ha cambiado, cuando estoy modificando una row, tengo que eliminar la misma de cassandra y sus relaciones, antes de poder agregarla la nueva.
				public.AddressesBook.Remove(oldAddressIdWithVcc)
			end
			-- Este atributo NO debe almacenarse en cassandra.
			addressBook.OldAddressInfo = nil

		end
		-- Obtengo el current Address
		local success, currentAddressesBook = public.AddressesBook.Get(addressIdWithVcc)
		if success then
			-- Si existe un current address (estoy modificando), tengo que eliminar las relaciones para luego actualizarlas
			for _, channel in ipairs(currentAddressesBook.Channels or {}) do
				RemoveAddressesBookRelations(keyspace, {AddressId = addressIdWithVcc, Channel = channel})
			end
		end
		-- Agrego la entidad
		helpers.SaveRows {
			keyspace = keyspace,
			columnFamily = "AddressesBook",
			rows = {
				[addressIdWithVcc] = addressBook
			}
		}
		-- Agrego las relaciones
		for _, channel in ipairs(addressBook.Channels or {}) do
			AddAddressesBookRelations(keyspace, {AddressId = addressIdWithVcc, Channel = channel})
		end
		return true
	end

	local success, result, err = with_connection_do(Save)
	logger[success and "info" or "warn"]("[FullContact::AddressesBook:Save] --> Saving address book result '%s'. err: '%s'.", json.encode(result or {}), json.encode(err or {}))
	return success, result, err
end

---
--
public.AddressesBook.Remove = function(addressId)
	assert(addressId)
	addressId = addressId:upper()
	local success = false
	logger.info("[FULLCONTACT.AddressesBook:Remove] Wait moment please ... Removing Addresses Book in Cassandra for '%s'.", addressId)

	local function Remove (keyspace)
		-- Si Valido si existe antes de remover
		local success, currentAddressesBook = public.AddressesBook.Get(addressId)
		if not success then
			return false, {err = "Addresses Book does not exist", addressId = addressId}
		end

		-- Elimino las Relaciones
		for _, channel in ipairs(currentAddressesBook.Channels or {}) do
			RemoveAddressesBookRelations(keyspace, {AddressId = addressId, Channel = channel})
		end

		-- Elimino la Entidad
		keyspace:RemoveColumn(addressId, "AddressesBook")

		return true
	end

	local success, err = with_connection_do(Remove)
	return success, err
end

-------------------------
-- Contacts Dialer
--------------------------
public.Dialer = {}
-------------------

public.Dialer.SaveContact = function (self, contact, options)
	--logger.info("[FULLCONTACT:SaveContact] - Saving contact '%s'.", helpers.JSON_Encode(contact))
	options = options or {}
	local elasticType = "CONTACTS_DIALER"
	return add_contact_search_engine(contact, elasticType, options)
end

public.Dialer.SaveContactsBatch = function (self, vcc, contacts, options)
	--logger.info("[FULLCONTACT:SaveContactsBatch] -> Saving '%s' contacts BATCH in vcc.", #contacts, vcc)
	--logger.debug("[FULLCONTACT:SaveContactsBatch] - Saving contacts BATCH '%s'.", helpers.JSON_Encode(contacts))
	options = options or {}

	for _, contact in ipairs(contacts) do
		contact.addresses = build_address_to_index(contact.addresses, vcc)
	end

	local success, result, detail = ES.Dialer:ValidateDupAndStoreContactsBatch(vcc, contacts, options)
	if not success then
		-- error
		local err = result
		return false, err, detail
	end
	-- ok
	return true, result, detail
end

--[[

]]--
public.Dialer.MigrateContactsBatch = function (self, vcc, contacts, options)
	--logger.info("[FULLCONTACT:MigrateContactsBatch] -> Migrating Allegro Contacts '%s' contacts BATCH in vcc.", #contacts, vcc)
	--logger.debug("[FULLCONTACT:MigrateContactsBatch] - Migrating Allegro contacts BATCH '%s'.", helpers.JSON_Encode(contacts))
	options = options or {}
	local contactsToMigrate = {}
	for _, contact in ipairs(contacts) do
		table.insert(contactsToMigrate, {
			action = "CREATE",
			contact = contact
		})
	end
	local success, result = ES.Dialer:MigrateContactsBatch(vcc, contactsToMigrate, options)
	if not success then
		-- error
		local err = result
		return false, err
	end
	-- ok
	return true, result
end

public.Dialer.SearchContacts = function (self, params)
	local vcc =  params.vcc or params.VCC
	--logger.info("[FULLCONTAC.Dialer:SearchContacts] - Searching Dialer Contacts in VCC '%s' using ELASTIC SEARCH: filter '%s', start '%s', rows per page '%s', sorted: '%s'.", vcc, helpers.JSON_Encode(params.filter), params.startRow, params.rowsPerPage, (params.sort and helpers.JSON_Encode(params.sort) or "NONE"))

	local query = {
		filter = params.filter,
		startRow = params.startRow,
		rowsPerPage = params.rowsPerPage,
		VCC = vcc
	}

	local success, response = ES.Dialer:SearchContacts(query)
	if not success then
		local err = response
		return false, err
	end

	return true, response
end

public.Dialer.GetContact = function (self, contactId)
	--logger.info("[FULLCONTACT:GetContact] - Get contact '%s'.", contactId)
	assert(type(contactId) == "string")
	local elasticType = "CONTACTS_DIALER"
	local simpleId, contactVCC = helpers.SeparateVCC(contactId)
	vcc = vcc or contactVCC
	local fullId = helpers.AddVCC(simpleId, vcc)
	return ES.Dialer:GetContact(fullId, vcc)
end

public.Dialer.GetContactByExternalId = function (self, externalId, vcc)
	--logger.info("[FULLCONTACT:GetContactByExternalId] - Get contact by external id '%s', vcc: %s.", externalId, vcc)
	assert(type(externalId) == "string")
	assert(type(vcc) == "string")
	return ES.Dialer:GetContactByExternalId(externalId, vcc)
end

---
--
public.Dialer.GetImportFormat = function(importFormatId, rowCount, ksOE)
	assert(type(importFormatId) == "string")
	rowCount = rowCount or 10000

	importFormatId = importFormatId:lower()
	--logger.info("[FULLCONTACT.Dialer:GetImportFormat] Wait moment please ... Retrieving Import Format Id '%s'.", importFormatId)

	local importFormatInfo = helpers.GetAllColumns {
		keyspace = ksOE or "OutboundEngine",
		columnFamily = "ContactImportFormat",
		key = importFormatId,
		rowCount = rowCount
	}
	--logger.debug("[FULLCONTACT.Dialer:GetImportFormat] Retrieved Import Format from Cassandra '%s'.", helpers.JSON_Encode(importFormatInfo or {}))

	if helpers.TableIsEmpty(importFormatInfo) then
		logger.info("[FULLCONTACT:Dialer:GetImportFormat] Import format '%s' does not exist", importFormatId)
		return false
	end
	--logger.info("[FULLCONTACT.Dialer:GetImportFormat] Is Found the Import Format Id '%s'.", importFormatId)

	return true, importFormatInfo
end

public.Dialer.SaveImportFormat = function (importFormatInfo, ksOE)
	assert(type(importFormatInfo) == "table")
	assert(type(importFormatInfo.Id) == "string")
	assert(type(importFormatInfo.VCC) == "string")
	assert(type(importFormatInfo.ContactImportFormatFields) == "table")
	assert(#importFormatInfo.ContactImportFormatFields > 0, "At least 1 field is required in the import format")
	local importFormatId = helpers.AddVCC(importFormatInfo.Id, importFormatInfo.VCC):lower()
	--logger.info("[FULLCONTACT.Dialer:SaveImportFormat] Wait moment please ... Saving Import format Id '%s' in Cassandra.", importFormatId)
	--logger.info("[FULLCONTACT.Dialer:SaveImportFormat] ---> Saving Import format Data '%s' in Cassandra.", helpers.JSON_Encode(importFormatInfo))
	local function Save (keyspace)
		-- Busco si ya existe
		local alreadyExist, currentImportInfo = public.Dialer.GetImportFormat(importFormatId, nil, ksOE)
		if alreadyExist then
			importFormatInfo.LastModificationDate = os.time()
		else
			importFormatInfo.CreationDate = os.time()
		end
		-- Agrego / actualizo la entidad
		helpers.SaveRows {
			keyspace = keyspace,
			columnFamily = "ContactImportFormat",
			rows = {
				[importFormatId] = importFormatInfo
			}
		}
		return true
	end
	if ksOE then
		return Save(ksOE)
	end
	local success, err = with_connection_do("OutboundEngine", Save)
	return success, importFormatId, err
end

---
--
public.Dialer.RemoveImportFormat = function(importFormatId, ksOE)
	assert(type(importFormatId) == "string")
	importFormatId = importFormatId:lower()
	local success = false
	--logger.info("[FULLCONTACT.Dialer:RemoveImportFormat] Wait moment please ... Removing Dialer Import Format from Cassandra for '%s'.", importFormatId)
	local function Remove (keyspace)
		-- valido si existe antes de remover
		local success, importFormatInfo = public.Dialer.GetImportFormat(importFormatId, nil, keyspace)
		if not success then
			logger.error("[FULLCONTACT:Dialer:RemoveImportFormat] Unable to delete Import format '%s'. Does not exist", importFormatId)
			return false, {err = "Import Format does not exist", importFormatId = importFormatId}
		end
		-- elimino la Entidad
		keyspace:RemoveColumn(importFormatId, "ContactImportFormat")
		return true
	end
	if ksOE then
		return Remove(ksOE)
	end
	local success, err = with_connection_do("OutboundEngine", Remove)
	return success, err
end

---
--
public.Dialer.ListImportFormat = function(options, ksOE)
	options = options or {}

	--logger.info("[FULLCONTACT.Dialer:ListImportFormat] Wait moment please ... Listing Import Formats from Cassandra for '%s'.", helpers.JSON_Encode(options))

	local releaseConnection = false
	if not ksOE then
		ksOE = m_datasource.cassandra:CassandraConnection("OutboundEngine")
		releaseConnection = true
	end
	local allImportFormats = helpers.GetAllColumns({
		keyspace = ksOE,
		columnFamily = "ContactImportFormat",
		startKey = "",
		endKey = "",
		rows_per_chunk = options.rows_per_chunk or 1000,
		sleep_between_chunk_sec = options.sleep_between_chunk_sec or nil
	}) or {}
	if releaseConnection then
		ksOE:Close()
	end

	local vcc = options.vcc and utf8.lower(options.vcc) or nil
	local importFormats = {}
	--local chunkTrace = 500
	--local i = 0

	for idAtVCC, importFormat in pairs(allImportFormats) do
		local id, importFormatVCC = helpers.SeparateVCC(idAtVCC)
		importFormatVCC = utf8.lower(importFormatVCC or "")
		--logger.debug("[FULLCONTACT.Dialer:ListImportFormat] importFormatVCC %s  vcc '%s'... wait moment.", importFormatVCC, vcc)
		if vcc == importFormatVCC then
			-- i = i + 1
			-- if i == chunkTrace then
			-- 	chunkTrace = chunkTrace + 500
			-- 	--logger.debug("[FULLCONTACT.Dialer:ListImportFormat] Loaded %s Import Formats of many for VCC '%s'... wait moment.", i, vcc)
			-- end
			table.insert(importFormats, importFormat)
		end
	end

	--logger.info("[FULLCONTACT.Dialer:ListImportFormat] Finished List. Total %s Import Formats from Cassandra.", i)
	--logger.debug("[FULLCONTACT.Dialer:ListImportFormat] Finished List Import Formats data: %s.", helpers.JSON_Encode(importFormats))

	return true, importFormats
end

---
--
public.Dialer.GetImportation = function(importId, rowCount, ksOE)
	assert(type(importId) == "string")
	rowCount = rowCount or 1000

	importId = importId:lower()
	--logger.debug("[FULLCONTACT.Dialer:GetImportation] Wait moment please ... Retrieving Contact Import Id '%s'.", importId)

	local contactImportInfo = helpers.GetAllColumns {
		keyspace = ksOE or "OutboundEngine",
		columnFamily = "ContactImport",
		key = importId,
		rowCount = rowCount
	}
	--logger.debug("[FULLCONTACT.Dialer:GetImportation] Retrieved Contact Import from Cassandra '%s'.", helpers.JSON_Encode(contactImportInfo or {}))

	if helpers.TableIsEmpty(contactImportInfo) then
		logger.info("[FULLCONTACT:Dialer:GetImportation] Contact Import '%s' does not exist", importId)
		return false
	end
	--logger.debug("[FULLCONTACT.Dialer:GetImportation] Is Found the Contact Import Id '%s'.", importId)

	return true, contactImportInfo
end

---
--
public.Dialer.GetImportFailedContacts = function(importId, detailId)
	assert(type(importId) == "string")
	assert(type(detailId) == "string")

	importId = importId:lower()
	detailId = detailId:lower()

	--logger.info("[FULLCONTACT.Dialer:GetImportFailedContacts] Wait moment please ... Retrieving Contact Import Failed Contacts. Id '%s' detailId: '%s'.", importId, detailId)

	local importDetailsFailedContacts = helpers.GetAllColumns {
		keyspace = "OutboundEngine",
		columnFamily = "ContactImportFailedContacts",
		superColumns = true,
		key = importId
	}

	--logger.debug("[FULLCONTACT.Dialer:GetImportFailedContacts] Retrieved Contact Import Failed Details from Cassandra '%s'.", helpers.JSON_Encode(importDetailsFailedContacts or {}))

	if helpers.TableIsEmpty(importDetailsFailedContacts) then
		--logger.info("[FULLCONTACT:Dialer:GetImportFailedContacts] Importation '%s' no has failed contacts.", importId)
		return false
	end

	local failedContacts = importDetailsFailedContacts[detailId]
	if not failedContacts then
		--logger.info("[FULLCONTACT:Dialer:GetImportFailedContacts] Importation '%s' no hast failed contacts detailId: %s.", importId, detailId)
		return false
	end

	--logger.info("[FULLCONTACT.Dialer:GetImportFailedContacts] Is Found the Contact Import Failed Contacts. Id '%s' detailId: '%s'.", importId, detailId)
	return true, failedContacts
end

---
--
public.Dialer.SaveContactImport = function (self, contactImportInfo, status, updateOnlyRealtime, failedContactsList, keyspace)
	assert(type(contactImportInfo) == "table")
	assert(type(contactImportInfo.ID) == "string")
	assert(type(status) == "table")

	local releaseConnection = false
	if not keyspace then
		keyspace = m_datasource.cassandra:CassandraConnection("OutboundEngine")
		releaseConnection = true
	end
	assert(keyspace)

	local importId = helpers.AddVCC(contactImportInfo.ID, contactImportInfo.VCC):lower()

	--logger.info("[FULLCONTACT.Dialer:SaveContactImport] Wait moment please ... Saving Contact Import Id '%s' in Cassandra, updateOnlyRealtime: %s.", importId, updateOnlyRealtime)
	--logger.debug("[FULLCONTACT.Dialer:SaveContactImport] ---> Saving Contact Import Data '%s' in Cassandra.", helpers.JSON_Encode(contactImportInfo))
	--logger.debug("[FULLCONTACT.Dialer:SaveContactImport] ---> Saving Contact Import status '%s' in Cassandra.", helpers.JSON_Encode(status))

	failedContactsList = failedContactsList or {}
	local detailId

	local function Save ()
		local contactImport = {
			ID         = contactImportInfo.ID,
			VCC        = contactImportInfo.VCC,
			DBProvider = contactImportInfo.DBProvider,
			FormatId   = contactImportInfo.FormatId,
			Status     = status,
			Updates    = 0,
			Detail     = {}
		}

		local function NewDetail (startDate)
			-- Linea del detalle
			detailId = uuid.new()
			local detail = {
				ImportId  = contactImportInfo.ID,
				VCC       = contactImportInfo.VCC,
				DetailId  = detailId,
				StartDate = startDate,
				UserId    = contactImportInfo.UserId,
				FormatId  = contactImport.FormatId,
				Status    = status.status,
				Total     = status.total,
				Added     = status.added,
				Updated   = status.updated,
				Deleted   = status.deleted,
				Warnings  = status.warnings,
				Errors    = status.errors
			}
			table.insert(contactImport.Detail, detail)
			-----------------
			-- Actualizo el current process en REDIS
			--   Creo el status del proceso en redis.
			local currentProcessBatch = {
				Id             = detailId,
				ImportId       = contactImport.ID,
				VCC            = contactImportInfo.VCC,
				StartDate      = LuaOrm.Date():toString(),
				User           = contactImportInfo.UserId,
				Status         = status.status or "RUNNING",
				Action         = status.action or "START PROCESS",
				ProcessCounter = 1,
				SubAction      = "...",
				Detail         = detail
			}
			helpers.dialer.UpdateVCCStatusProcessBatch(currentProcessBatch)
			--
			return detail
		end

		-- Busco si ya existe
		local alreadyExist, currentContactImportInfo = public.Dialer.GetImportation(importId, 1, keyspace)
		if alreadyExist then
			-- Actualizando
			contactImport.LastModificationDate = os.time()
			contactImport.Detail = currentContactImportInfo.Detail
			local currentDetail = contactImport.Detail[#contactImport.Detail]
			if currentDetail.Status ~= "RUNNING" and currentDetail.Status ~= "DELETING" then
				-- Nueva ejecucion / nuevo detalle
				NewDetail(contactImport.LastModificationDate)
			else
				-- update current detail
				detailId = currentDetail.DetailId
				currentDetail.UpdateDate = os.time()
				currentDetail.Added = status.added
				currentDetail.Updated = status.updated
				currentDetail.Errors = status.errors
				currentDetail.Warnings = status.warnings
				currentDetail.Deleted = status.deleted
				currentDetail.Status = status.status
				contactImport.Detail[#contactImport.Detail] = currentDetail
				-- obtengo el current process de REDIS
				local encodeCurrentProcessBatch = helpers.dialer.GetVCCStatusProcessBatch(contactImportInfo.VCC)
				if encodeCurrentProcessBatch then
					local currentProcessBatch = helpers.JSON_Decode(encodeCurrentProcessBatch)
					if currentDetail.Status == "COMPLETE" then
						helpers.dialer.DeleteVCCProcessBatch(currentProcessBatch.Id, contactImportInfo.VCC)
					else
						-- Actualizo el current process en REDIS
						currentProcessBatch.Action = status.action or "PROCESSING"
						currentProcessBatch.UpdatedDate = LuaOrm.Date():toString()
						currentProcessBatch.Detail = currentDetail
						currentProcessBatch.ProcessCounter = currentProcessBatch.ProcessCounter + 1
						--
						helpers.dialer.UpdateVCCStatusProcessBatch(currentProcessBatch)
					end
				end
			end
		else
			-- Nuevo
			contactImport.UserId = contactImportInfo.UserId
			contactImport.CreationDate = os.time()
			NewDetail(contactImport.CreationDate)
		end

		contactImport.Updates = #contactImport.Detail

		-- Agrego / actualizo la entidad en CASSANDRA
		if not updateOnlyRealtime then
			public.Dialer.SaveContactImportInDB(importId, contactImport, keyspace)
			if #failedContactsList > 0 then
				local contactImportFailedContacts = {}
				contactImportFailedContacts[detailId] = failedContactsList
				helpers.SaveRows {
					keyspace = keyspace,
					superColumns = true,
					columnFamily = "ContactImportFailedContacts",
					rows = {
						[importId] = contactImportFailedContacts
					}
				}
			end
		end

		--logger.debug("[FULLCONTACT.Dialer:SaveContactImport] ---> Success saved Contact Import '%s' in Cassandra.", importId)
		return true
	end

	local success, err = Save()
	if releaseConnection then
		--logger.debug("[FULLCONTACT.Dialer:SaveContactImport]-------> RELEASE Cassandra CONNECTION !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!.")
		keyspace:Close()
	end

	return success, importFormatId, err
end

---
--
public.Dialer.SaveContactImportInDB = function(importIdWithVCC, contactImport, ksOE)
	assert(type(importIdWithVCC) == "string")
	assert(type(contactImport) == "table")
	local releaseConnection = false
	if not ksOE then
		ksOE = m_datasource.cassandra:CassandraConnection("OutboundEngine")
		releaseConnection = true
	end
	helpers.SaveRows {
		keyspace = ksOE,
		columnFamily = "ContactImport",
		rows = {
			[importIdWithVCC] = contactImport
		}
	}
	if releaseConnection then
		ksOE:Close()
	end
	return true
end

---
--
public.Dialer.RemoveContactImportFromDB = function(importIdWithVCC, ksOE)
	assert(type(importIdWithVCC) == "string")
	local releaseConnection = false
	if not ksOE then
		ksOE = m_datasource.cassandra:CassandraConnection("OutboundEngine")
		releaseConnection = true
	end
	-- Elimino de CASSANDRA la Importación.
	ksOE:RemoveColumn(importIdWithVCC, "ContactImport")
	-- Elimino de CASSANDRA los contactos fallidos de la Importación.
	ksOE:RemoveColumn(importIdWithVCC, "ContactImportFailedContacts")
	if releaseConnection then
		ksOE:Close()
	end
	return true
end


---
--
public.Dialer.ListImportations = function(options, ksOE)
	options = options or {}

	local releaseConnection = false
	if not ksOE then
		ksOE = m_datasource.cassandra:CassandraConnection("OutboundEngine")
		releaseConnection = true
	end

	--logger.info("[FULLCONTACT.Dialer:ListImportations] Wait moment please ... Listing Imports from Cassandra for '%s'.", helpers.JSON_Encode(options))

	local keyspace = m_datasource.cassandra:CassandraConnection("OutboundEngine")
	local importations = helpers.GetAllColumns({
		keyspace = ksOE,
		columnFamily = "ContactImport",
		startKey = "",
		endKey = "",
		rows_per_chunk = options.rows_per_chunk or 1000,
		sleep_between_chunk_sec = options.sleep_between_chunk_sec or nil
	}) or {}
	if releaseConnection then
		ksOE:Close()
	end

	local vcc = options.vcc and utf8.lower(options.vcc) or nil
	local imports = {}
	--local chunkTrace = 500
	local i = 0

	for idAtVCC, import in pairs(importations) do
		local id, importVCC = helpers.SeparateVCC(idAtVCC)
		importVCC = utf8.lower(importVCC or "")
		--logger.debug("[FULLCONTACT.Dialer:ListImportations] importVCC %s  vcc '%s'... wait moment.", importVCC, vcc)
		if vcc == importVCC then
			--i = i + 1
			--if i == chunkTrace then
				--chunkTrace = chunkTrace + 500
				--logger.debug("[FULLCONTACT.Dialer:ListImportations] Loaded %s Imports of many for VCC '%s'... wait moment.", i, vcc)
			--end
			table.insert(imports, import)
		end
	end

	--logger.info("[FULLCONTACT.Dialer:ListImportations] Finished List. Total %s Imports from Cassandra.", i)
	--logger.debug("[FULLCONTACT.Dialer:ListImportations] Finished List Imports data: %s.", helpers.JSON_Encode(imports))

	return true, imports
end

---
--
public.Dialer.CanRemoveOrDepurateImport = function(importId, isBlackList, ksOE)
	assert(type(importId) == "string")
	importId = importId:lower()
	-- valido si existe antes de remover
	local success
	if isBlackList then
		success, importationInfo = public.BlackList.GetImportation(importId, nil, ksOE)
	else
		success, importationInfo = public.Dialer.GetImportation(importId, nil, ksOE)
	end
	if not success then
		logger.error("[FULLCONTACT:Dialer:CanRemoveOrDepurateImport] Unable to delete/depurate Importation '%s'. Does not exist", importId)
		return false, {err = "IMPORTATION DOES NOT EXIST", importId = importId, type = "-1"}
	end
	-- obtengo el current process de REDIS
	--
	-- 1- Valido que NO exista ningun otro proceso de batch en curso (exportacion, alta, baja)
	local isAnyProcessBatch, currentImportInProgress = helpers.dialer.IsAnyProcessBatchInProgressInVCC(importationInfo.VCC)
	if isAnyProcessBatch then
		-- Error.
		-- La mimsa importacion se encuentra en uso
		local err = {
			importId = helpers.SeparateVCC(importId),
			result = "THERE CURRENT IMPORT IN PROGRESS",
			reason = "Unable to delete/depurate Import. There current import in progress",
			type = "-2"
		}
		--LogDebug("1 ---> currentImportInProgress: %s", currentImportInProgress)
		--LogDebug("2 ---> currentImportInProgress.ImportId: %s", (currentImportInProgress or {}).ImportId)
		if currentImportInProgress and currentImportInProgress.ImportId then
			local importInProgress = helpers.SeparateVCC(currentImportInProgress.ImportId):lower()
			if importInProgress ~= err.importId then
				-- cambio el tipo de error ya que es otra importacion la que se encuentra en curso
				err.result = "THERE ANOTHER IMPORT IN PROGRESS"
				err.reason = "Unable to delete/depurate Import. There another import in progress"
				err.type = "-11"
				err.importInProgress = currentImportInProgress.ImportId
			end
		end
		logger.warn("[FULLCONTACT.Dialer:CanRemoveOrDepurateImport] Unable to delete/depurate Import %s. Dialer Contact Import Error. %s", importId, helpers.JSON_Encode(err))
		-- Is Error
		return false, err
	end
	-- ok - se puede remover o depurar
	return true, importationInfo
end

---
--
public.Dialer.CanRemoveContact = function(contact, vcc)
	assert(type(contact) == "table")
	--logger.debug("[FULLCONTACT.Dialer:CanRemoveContact] Validate removing contact: '%s' vcc: '%s'. contact: %s", contact.Id, vcc, json.encode(contact))
	if type(contact.BatchId) == "string" and #contact.BatchId > 0 then
		-- no se puede borrar el contacto, el mismo se encuentra en un batch
		local msg = "Unable to delete/depurate Import. There contacts are found in some batches."
		local err = {
			result = "ERROR",
			result = "THERE CONTACTS ARE FOUND IN SOME BATCHES",
			reason = msg,
			contactId = contact.Id,
			batchId = contact.BatchId,
			type = "-10"
		}
		--logger.warn("[FULLCONTACT.Dialer:CanRemoveContact]---> Unable to remove contact: '%s' vcc: '%s'. The contact is in the batch: %s", contact.Id, vcc, contact.BatchId)
		return false, err
	end
	-- ok - se puede borrar el contacto
	return true
end

---
--
public.Dialer.RemoveImport = function(importId, userId, pKsOE)
	assert(type(importId) == "string")
	assert(IMPORTATION_TYPE == "BATCH", "unrecognized IMPORTATION_TYPE" .. tostring(IMPORTATION_TYPE))

	importId = importId:lower()
	userId = userId or "ANONYMUS"

	logger.info("[FULLCONTACT.Dialer:RemoveImport] Wait moment please ... Removing Dialer IMPORTATION '%s'. IMPORTATION_TYPE '%s'.", importId, IMPORTATION_TYPE)

	local ksOE = pKsOE or m_datasource.cassandra:CassandraConnection("OutboundEngine")
	local importationInfo
	local function CanRemoveImportation ()
		-- valido si existe antes de remover
		local success
		success, importationInfo = public.Dialer.CanRemoveOrDepurateImport(importId, false, ksOE)
		return success, importationInfo
	end

	local statusInfo = {
		status = "DELETING",
		action = "START PROCESS",
		added = 0,
		updated = 0,
		errors = 0,
		deleted = 0
	}

	local function UpdateVCCStatusProcessBatch (updateOnlyRealtime)
		public.Dialer:SaveContactImport(importationInfo, statusInfo, updateOnlyRealtime, nil, ksOE)
	end

	local deletedContacts = {}

	local function RollBackContacts (errRemoveContact)
		-- CANCELO si es que existe al menos 1 contacto que no se puede remover.
		statusInfo.status = "COMPLETE"
		statusInfo.action = "ENDED DELETING"
		UpdateVCCStatusProcessBatch()

		local err = errRemoveContact or {code = "UNABLE TO REMOVE IMPORTATION", type = "-3", reason = "there are contacts that cannot be removed"}
		logger.warn("[FULLCONTACT.Dialer:RemoveImport]---> Unable to remove importation: '%s' vcc: '%s'. %s.", importId, importationInfo.VCC, json.encode(err))

		local contactsToRollback = {}
		for _, deletedContact in pairs(deletedContacts or {}) do
			local rollBack = {
				action  = "CREATE",
				contact = deletedContact
			}
			--logger.debug("[FULLCONTACT.Dialer:RemoveImport.UpdateCurrentContact] ---> rollBack: '%s'.", helpers.JSON_Encode(rollBack))
			table.insert(contactsToRollback, rollBack)
		end
		local success, result = ES.Dialer:UpdateContactsBatch(importationInfo.VCC, contactsToRollback)
		return false, err
	end

	local function RemoveContactsImport ()
		require("socket").sleep(1) -- hay que esperar un tiempo determinado (1 segundo) para que el delete se haga efectivo y luego realizar la siguiente consulta.
		local success, contacts = public.Dialer.LoadPaginContactsOfImportation(importId)
		if not success or (success and #contacts == 0) then
			logger.info("[FULLCONTACT.Dialer:RemoveImport] No more contacts in Importation '%s' to remove.", importId)
			return true, {code = "NO MORE CONTACTS"}
		end

		-- 2 - valido si los contactos pueden ser removidos.
		contacts = contacts or {}
		local canRemovePaginContacts = true
		local errRemoveContact
		for _, contact in ipairs(contacts) do
			if not deletedContacts[contact.Id] then
				local canRemove, err = public.Dialer.CanRemoveContact(contact, importationInfo.VCC)
				if not canRemove then
					canRemovePaginContacts = false
					errRemoveContact = err
					break
				end
			--else
				--logger.debug("[FULLCONTACT.Dialer:RemoveImport] --> Already deleted contact '%s'.", contact.Id)
			end
		end

		if not canRemovePaginContacts then
			-- CANCELO si es que existe al menos 1 contacto que no se puede remover.
			return RollBackContacts(errRemoveContact)
		end

		-- 3 - Elimino de ES los contactos
		local updateStatusCounter = m_configValues.importation.updateStatusCounter
		local counter = 0
		local contactsToRemove = contacts

		if IMPORTATION_TYPE == "BATCH" then
			-- BATCH (Muchos)
			--
			local success, result, detail = ES.Dialer:RemoveContactsBatch(importationInfo.VCC, contactsToRemove)
			-- logger.debug("[FULLCONTACT:RemoveImport] 1- success '%s'.", success)
			-- logger.debug("[FULLCONTACT:RemoveImport] 2- result  '%s'.", json.encode(result))
			-- logger.debug("[FULLCONTACT:RemoveImport] 3- detail  '%s'.", json.encode(detail))
			if success then
				-- 2.
				--
				-- if detail.delete and (detail.delete.deleted or 0) < #contactsToRemove then
				-- 	for i, contact in ipairs(contactsToRemove) do
				-- 		deletedContacts[contact.Id] = contact
				-- 	end
				-- 	return RollBackContacts()
				-- end
			else
				-- 3. si es error, actualiza los contadores (errors + los invalid para descargar el CSV)
				--
				return RollBackContacts()
			end
			for i, contact in ipairs(contactsToRemove) do
				counter = counter + 1
				if counter == updateStatusCounter then
					statusInfo.status = "DELETING"
					statusInfo.action = "DELETED CONTACTS: " .. tostring(i)
					UpdateVCCStatusProcessBatch()
					counter = 0
				end
				deletedContacts[contact.Id] = contact
			end
		else
			assert(false, "DEPRECATED " .. tostring(IMPORTATION_TYPE))
		end
		-- ok
		if #contacts < ROWS_PER_PAGE then
			-- more contacts
			--logger.info("[FULLCONTACT.Dialer:RemoveImport] -- No more contacts in Importation '%s' to remove.", importId)
			return true, {code = "NO MORE CONTACTS"}
		end
		-- MORE CONTACTS
		return true, {code = "MORE CONTACTS"}
	end

	local function StartRemoveImport ()
		local success, err = CanRemoveImportation()
		if not success then
			return false, err
		end

		-- 1 - Obtengo los contactos de la importación.
		UpdateVCCStatusProcessBatch(true)
		local canRemoveImportation = false
		local currentLoadedRows = 0
		local err
		while true do
			currentLoadedRows = currentLoadedRows + ROWS_PER_PAGE
			local success, result = RemoveContactsImport()
			if success then
				if result.code == "NO MORE CONTACTS" then
					--logger.debug("[FULLCONTACT.Dialer:RemoveImport] 1 OK ---> NO MORE CONTACTS. total loaded: %s", currentLoadedRows)
					canRemoveImportation = true
					break
				--elseif result.code == "MORE CONTACTS" then
					--logger.debug("[FULLCONTACT.Dialer:RemoveImport] 2 OK ---> MORE CONTACTS. loaded: %s", currentLoadedRows)
				--else
					--logger.debug("[FULLCONTACT.Dialer:RemoveImport] 3 OK ---> Unrecognized code: %s. loaded: %s", result.code, currentLoadedRows)
				end
			else
				--logger.debug("[FULLCONTACT.Dialer:RemoveImport] 4 FAIL ---> Unable to delete contacts, code: %s. loaded: %s", result.code, currentLoadedRows)
				canRemoveImportation = false
				err = result or {err = "UNABLE TO COMPLETE DELETE CONTACTS OF IMPORTAION"}
				err.importId = importId
				err.type = err.type or "-4"
				break
			end
			if currentLoadedRows >= (MAXIMUM_NUMBER_OF_ROWS_IN_DIALER_IMPORT + ROWS_PER_PAGE) then
				canRemoveImportation = true
				--logger.debug("[FULLCONTACT.Dialer:RemoveImport] 5 FAIL ---> Unable to delete contacts: OVERFLOW. loaded: %s", currentLoadedRows)
				break
			end
		end

		if not canRemoveImportation then
			-- ERROR
			return false, err
		end
		-- OK
		--
		-- 4 - Actualizo el estado en REDIS.
		statusInfo.status = "COMPLETE"
		statusInfo.action = "ENDED DELETING"
		UpdateVCCStatusProcessBatch(true)
		-- 5 - Elimino de CASSANDRA la Importación y los contactos fallidos de la Importación.
		public.Dialer.RemoveContactImportFromDB(importId, ksOE)
		-- -- 6 - Elimino relacion de la importacion contra la importacion de blacklist.
		-- local currentBlacklists = importationInfo.Blacklists or {}
		-- for blackListId in pairs(currentBlacklists) do
		-- 	local success, err = public.BlackList.RemoveAssociatedContactsImports(blackListId, importId, ksOE) -- almaceno en la CF BlackListImport
		-- end
		return true
	end

	local success, resp = StartRemoveImport()
	if not pKsOE then
		ksOE:Close()
	end

	return success, resp
end

---
--
public.Dialer.DepurateImport = function(importId, userId)
	assert(type(importId) == "string")
	importId = importId:lower()
	userId = userId or "ANONYMUS"

	assert(IMPORTATION_TYPE == "BATCH", "unrecognized IMPORTATION_TYPE" .. tostring(IMPORTATION_TYPE))

	logger.info("[FULLCONTACT.Dialer:DepurateImport] Wait moment please ... Depurating Dialer IMPORTATION '%s'. . IMPORTATION_TYPE '%s'", importId, IMPORTATION_TYPE)

	local importationInfo
	local function CanDepurateImportation ()
		-- Si Valido si existe antes de remover
		local success
		success, importationInfo = public.Dialer.CanRemoveOrDepurateImport(importId)
		return success, importationInfo
	end

	local statusInfo = {
		status = "DELETING",
		action = "START PROCESS DEPURATE CONTACTS",
		added = 0,
		updated = 0,
		errors = 0,
		deleted = 0
	}

	local keyspace = m_datasource.cassandra:CassandraConnection("OutboundEngine")
	local function UpdateVCCStatusProcessBatch (updateOnlyRealtime)
		public.Dialer:SaveContactImport(importationInfo, statusInfo, updateOnlyRealtime, nil, keyspace)
	end

	local allDepuratedContacts = {}
	local contactsNotRemove = {}
	local passedContacts = {}

	local function DepurateContactsImport ()
		require("socket").sleep(1) -- hay que esperar un tiempo determinado (1 segundo) para que el delete se haga efectivo al realizar y luego realizar la siguiente consulta.
		local success, contacts = public.Dialer.LoadPaginContactsOfImportation(importId)
		if not success or (success and #contacts == 0) then
			--logger.info("[FULLCONTACT.Dialer:DepurateImport] No more contacts in Importation '%s' to depurate.", importId)
			return true, {code = "NO MORE CONTACTS"}
		end
		-- 2 - valido si los contactos pueden ser removidos.
		contacts = contacts or {}
		local contactsToRemove = {}
		local canRemovePaginContacts = true

		for _, contact in ipairs(contacts) do
			if not passedContacts[contact.Id] then
				local canRemove, err = public.Dialer.CanRemoveContact(contact, importationInfo.VCC)
				if canRemove then
					-- agrego a la lista de contactos que SI se pueden depurar.
					table.insert(contactsToRemove, contact)
				else
					-- agrego a la lista de contactos que NO se pueden depurar.
					table.insert(contactsNotRemove, contact.Id)
				end
			end
		end

		-- 3 - Elimino de ES los contactos
		local updateStatusCounter = m_configValues.importation.updateStatusCounter
		local counter = 0

		if IMPORTATION_TYPE == "BATCH" then
			-- BATCH (Muchos)
			--
			local success, result, detail = ES.Dialer:RemoveContactsBatch(importationInfo.VCC, contactsToRemove)
			-- logger.debug("[FULLCONTACT:RemoveImport] 1- success '%s'.", success)
			-- logger.debug("[FULLCONTACT:RemoveImport] 2- result  '%s'.", json.encode(result))
			-- logger.debug("[FULLCONTACT:RemoveImport] 3- detail  '%s'.", json.encode(detail))
			for i, contact in ipairs(contactsToRemove) do
				counter = counter + 1
				if counter == updateStatusCounter then
					statusInfo.status = "DELETING"
					statusInfo.action = "DEPURATED CONTACTS: " .. tostring(i)
					UpdateVCCStatusProcessBatch()
					counter = 0
				end
				passedContacts[contact.Id] = contact
				table.insert(allDepuratedContacts, contact)
			end
		else
			assert(false, "DEPRECATED " .. tostring(IMPORTATION_TYPE))
		end
		-- ok
		if #contacts < ROWS_PER_PAGE then
			-- more contacts
			--logger.info("[FULLCONTACT.Dialer:DepurateImport] -- No more contacts in Importation '%s' to remove.", importId)
			return true, {code = "NO MORE CONTACTS"}
		end
		-- MORE CONTACTS
		return true, {code = "MORE CONTACTS"}
	end

	local function StartDepurateImport ()
		local success, err = CanDepurateImportation()
		if not success then
			return false, err
		end
		-- 1 - Obtengo los contactos de la importación.
		UpdateVCCStatusProcessBatch(true)
		local currentLoadedRows = 0
		local err
		while true do
			currentLoadedRows = currentLoadedRows + ROWS_PER_PAGE
			local success, result = DepurateContactsImport()
			if success then
				if result.code == "NO MORE CONTACTS" then
					--logger.debug("[FULLCONTACT.Dialer:DepurateImport] 1 OK ---> NO MORE CONTACTS. total loaded: %s", currentLoadedRows)
					break
				--elseif result.code == "MORE CONTACTS" then
					--logger.debug("[FULLCONTACT.Dialer:DepurateImport] 2 OK ---> MORE CONTACTS. loaded: %s", currentLoadedRows)
				--else
					--logger.debug("[FULLCONTACT.Dialer:DepurateImport] 3 OK ---> Unrecognized code: %s. loaded: %s", result.code, currentLoadedRows)
				end
			else
				--logger.debug("[FULLCONTACT.Dialer:DepurateImport] 4 FAIL ---> Unable to delete contacts, code: %s. loaded: %s", result.code, currentLoadedRows)
				err = {err = "UNABLE TO COMPLETE DELETE CONTACTS OF IMPORTAION", importId = importId, type = "-4"}
				break
			end
			if currentLoadedRows >= (MAXIMUM_NUMBER_OF_ROWS_IN_DIALER_IMPORT + ROWS_PER_PAGE) then
				--logger.debug("[FULLCONTACT.Dialer:DepurateImport] 5 FAIL ---> Unable to delete contacts: OVERFLOW. loaded: %s", currentLoadedRows)
				break
			end
		end

		if #allDepuratedContacts == 0 then
			-- CANCELO si es que NO existen contactos que se puedan depurar.
			statusInfo.status = "COMPLETE"
			statusInfo.action = "ENDED DEPURATING CONTACTS - THERE ARE NO CONTACTS THAT CAN BE DEPURATE"
			UpdateVCCStatusProcessBatch()
			local err = {code = "UNABLE TO DEPURATE IMPORTATION", type = "-3", reason = "there are no contacts that can be depurate"}
			logger.info("[FULLCONTACT.Dialer:DepurateImport]---> Unable to depurate importation: '%s' vcc: '%s'. There are no contacts that can be depurate.", importId, importationInfo.VCC)
			return false, err
		end

		--
		statusInfo.status = "COMPLETE"
		statusInfo.action = "ENDED DEPURATING CONTACTS"
		UpdateVCCStatusProcessBatch()

		return true, {
			totalDepuratedContacts = #allDepuratedContacts,
			contacts = allDepuratedContacts
		}
	end

	local success, result = StartDepurateImport()
	keyspace:Close()
	return success, result
end

---
--
public.Dialer.LoadPaginContactsOfImportation = function(importId)
	assert(type(importId) == "string")
	importId = importId:lower()
	local rowsPerPage = ROWS_PER_PAGE

	--logger.info("[FULLCONTACT.Dialer:LoadPaginContactsOfImportation] Wait moment please ... Loading %d Contacts Dialer of Importation '%s'.", rowsPerPage, importId)

	local success, importationInfo = public.Dialer.GetImportation(importId)
	if not success then
		logger.error("[FULLCONTACT:Dialer:LoadPaginContactsOfImportation] Unable to retrieve Importation '%s'. Does not exist", importId)
		return false, {err = "IMPORTATION DOES NOT EXIST", importId = importId, type = "-1"}
	end

	local importIdWithoutVCC = helpers.SeparateVCC(importId)
	local function ExecuteQuery (startRow)
		local query = {
			filter = {
				ImportId = {
					exact  = true,
					values = {importIdWithoutVCC}
				}
			},
			vcc         = importationInfo.VCC,
			startRow    = startRow,
			rowsPerPage = rowsPerPage
		}
		local success, response = ES.Dialer:SearchContacts(query)
		--logger.debug("[FULLCONTACT:Dialer:LoadPaginContactsOfImportation] Loaded dialer contacts of importation '%s' vcc: '%s'. success: %s, response: %s", importId, importationInfo.VCC, success, helpers.JSON_Encode(response or {}))
		return success, response
	end

	local from = 0
	local success, response = ExecuteQuery(from)
	if not success then
		return false, response
	end
	local contacts = {}
	local responseContacts = response.contacts or {}
	for _, contact in ipairs(responseContacts) do
		table.insert(contacts, contact)
	end

	--logger.debug("[FULLCONTACT:Dialer:LoadPaginContactsOfImportation] Loading %d contacts dialer of importation '%s' vcc: '%s'.", #contacts, importId, importationInfo.VCC)

	return true, contacts
end

-------------------------
-- BlackList
--------------------------
public.BlackList = {}

-------------------
---
--
public.BlackList.GetImportation = function(importId, rowCount, ksOE)
	assert(type(importId) == "string")
	rowCount = rowCount or 1000

	importId = importId:lower()
	--logger.debug("[FULLCONTACT.BlackList:GetImportation] Wait moment please ... Retrieving Contact Import Id '%s'.", importId)

	local contactImportInfo = helpers.GetAllColumns {
		keyspace = ksOE or "OutboundEngine",
		columnFamily = "BlackListImport",
		key = importId,
		rowCount = rowCount
	}
	--logger.debug("[FULLCONTACT.BlackList:GetImportation] Retrieved Contact Import from Cassandra '%s'.", helpers.JSON_Encode(contactImportInfo or {}))

	if helpers.TableIsEmpty(contactImportInfo) then
		logger.info("[FULLCONTACT:BlackList:GetImportation] Blacklist Import '%s' does not exist", importId)
		return false
	end
	--logger.debug("[FULLCONTACT.BlackList:GetImportation] Is Found the Contact Import Id '%s'.", importId)

	return true, contactImportInfo
end

---
--
public.BlackList.Save = function(blackListId, blackListImportInfo, keyspace)
	helpers.SaveRows {
		keyspace = keyspace,
		columnFamily = "BlackListImport",
		rows = {
			[blackListId] = blackListImportInfo
		}
	}
end

---
--
public.BlackList.ListImportations = function(options)
	options = options or {}

	--logger.info("[FULLCONTACT.BlackList:ListImportations] Wait moment please ... Listing Imports from Cassandra for '%s'.", helpers.JSON_Encode(options))

	local keyspace = m_datasource.cassandra:CassandraConnection("OutboundEngine")
	local importations = helpers.GetAllColumns({
		keyspace = keyspace,
		columnFamily = "BlackListImport",
		startKey = "",
		endKey = "",
		rows_per_chunk = options.rows_per_chunk or 1000,
		sleep_between_chunk_sec = options.sleep_between_chunk_sec or nil
	}) or {}
	keyspace:Close()

	local vcc = options.vcc and utf8.lower(options.vcc) or nil
	local addressType = options.addressType and utf8.upper(options.addressType) or nil
	local imports = {}

	for idAtVCC, import in pairs(importations) do
		local id, importVCC = helpers.SeparateVCC(idAtVCC)
		importVCC = utf8.lower(importVCC or "")
		importAddressType = utf8.upper(import.AddressType or "")
		--logger.debug("[FULLCONTACT.BlackList:ListImportations] importVCC %s  vcc '%s' importAddressType '%s'... wait moment.", importVCC, vcc, importAddressType)
		if vcc == importVCC and (not addressType or addressType == importAddressType) then
			table.insert(imports, import)
		end
	end

	--logger.info("[FULLCONTACT.BlackList:ListImportations] Finished List. Total %s Imports from Cassandra.", i)
	--logger.debug("[FULLCONTACT.BlackList:ListImportations] Finished List Imports data: %s.", helpers.JSON_Encode(imports))

	return true, imports
end

---
--
public.BlackList.SaveImportation = function (self, blackListImportInfo, status, updateOnlyRealtime, failedAddressesList, ksOE)
	assert(type(blackListImportInfo) == "table")
	assert(type(blackListImportInfo.ID) == "string")
	assert(type(status) == "table")

	local releaseConnection = false
	if not ksOE then
		ksOE = m_datasource.cassandra:CassandraConnection("OutboundEngine")
		releaseConnection = true
	end
	assert(ksOE)

	local importId = helpers.AddVCC(blackListImportInfo.ID, blackListImportInfo.VCC):lower()

	--logger.info("[FULLCONTACT.BlackList:SaveImportation] Wait moment please ... Saving BlackList Import Id '%s' in Cassandra, updateOnlyRealtime: %s.", importId, updateOnlyRealtime)
	--logger.debug("[FULLCONTACT.BlackList:SaveImportation] 1---> Saving BlackList Import Data '%s' in Cassandra.", helpers.JSON_Encode(blackListImportInfo))
	--logger.debug("[FULLCONTACT.BlackList:SaveImportation] 2---> Saving BlackList Import status '%s' in Cassandra.", helpers.JSON_Encode(status))

	failedAddressesList = failedAddressesList or {}

	--logger.debug("[FULLCONTACT.BlackList:SaveImportation] 3 ------------------> Saving BlackList failedAddressesList: %s.", helpers.JSON_Encode(failedAddressesList))

	local detailId

	local function Save ()
		local blackListImport = {
			ID          = blackListImportInfo.ID,
			VCC         = blackListImportInfo.VCC,
			Status      = status,
			AddressType = blackListImportInfo.AddressType,
			Description = blackListImportInfo.Description,
			Updates     = 0,
			Detail      = {}
		}

		local function NewDetail (startDate)
			-- Linea del detalle
			detailId = uuid.new()
			local detail = {
				ImportId  = blackListImportInfo.ID,
				VCC       = blackListImportInfo.VCC,
				DetailId  = detailId,
				StartDate = startDate,
				UserId    = blackListImportInfo.UserId,
				Status    = status.status,
				Total     = status.total,
				Added     = status.added,
				Updated   = status.updated,
				Deleted   = status.deleted,
				Errors    = status.errors,
				Warnings  = status.warnings
			}
			table.insert(blackListImport.Detail, detail)
			-----------------
			-- Actualizo el current process en REDIS
			--   Creo el status del proceso en redis.
			local currentProcessBatch = {
				Id             = detailId,
				ImportId       = blackListImport.ID,
				VCC            = blackListImportInfo.VCC,
				StartDate      = LuaOrm.Date():toString(),
				User           = blackListImportInfo.UserId,
				Status         = status.status or "RUNNING",
				Action         = status.action or "START PROCESS",
				ProcessCounter = 1,
				SubAction      = "...",
				Detail         = detail
			}
			helpers.dialer.UpdateVCCStatusProcessBatch(currentProcessBatch)
			--
			return detail
		end

		-- Busco si ya existe
		local alreadyExist, currentContactImportInfo = public.BlackList.GetImportation(importId, 1, ksOE)
		if alreadyExist then
			-- Actualizando
			blackListImport.LastModificationDate = os.time()
			blackListImport.Detail = currentContactImportInfo.Detail
			local currentDetail = blackListImport.Detail[#blackListImport.Detail]
			if currentDetail.Status ~= "RUNNING" and currentDetail.Status ~= "DELETING" then
				-- Nueva ejecucion / nuevo detalle
				NewDetail(blackListImport.LastModificationDate)
			else
				-- update current detail
				detailId = currentDetail.DetailId
				currentDetail.UpdateDate = os.time()
				currentDetail.Added = status.added
				currentDetail.Updated = status.updated
				currentDetail.Errors = status.errors
				currentDetail.Deleted = status.deleted
				currentDetail.Status = status.status
				blackListImport.Detail[#blackListImport.Detail] = currentDetail
				-- obtengo el current process de REDIS
				local encodeCurrentProcessBatch = helpers.dialer.GetVCCStatusProcessBatch(blackListImportInfo.VCC)
				if encodeCurrentProcessBatch then
					local currentProcessBatch = helpers.JSON_Decode(encodeCurrentProcessBatch)
					if currentDetail.Status == "COMPLETE" then
						helpers.dialer.DeleteVCCProcessBatch(currentProcessBatch.Id, blackListImportInfo.VCC)
					else
						-- Actualizo el current process en REDIS
						currentProcessBatch.Action = status.action or "PROCESSING"
						currentProcessBatch.UpdatedDate = LuaOrm.Date():toString()
						currentProcessBatch.Detail = currentDetail
						currentProcessBatch.ProcessCounter = currentProcessBatch.ProcessCounter + 1
						--
						helpers.dialer.UpdateVCCStatusProcessBatch(currentProcessBatch)
					end
				end
			end
		else
			-- Nuevo
			blackListImport.UserId = blackListImportInfo.UserId
			blackListImport.CreationDate = os.time()
			NewDetail(blackListImport.CreationDate)
		end

		blackListImport.Updates = #blackListImport.Detail

		-- Agrego / actualizo la entidad en CASSANDRA
		if not updateOnlyRealtime then
			-- almaceno en la CF BlackListImport
			public.BlackList.Save(importId, blackListImport, ksOE)
			-- almaceno en la CF BlackListFailedAddresses
			if #failedAddressesList > 0 then
				local blackListImportFailed = {}
				blackListImportFailed[detailId] = failedAddressesList
				helpers.SaveRows {
					keyspace = ksOE,
					superColumns = true,
					columnFamily = "BlackListFailedAddresses",
					rows = {
						[importId] = blackListImportFailed
					}
				}
			end
		end

		--logger.debug("[FULLCONTACT.BlackList:SaveImportation] ---> Success saved Blacklist Import '%s' in Cassandra.", importId)
		return true
	end

	local success, err = Save()
	if releaseConnection then
		--logger.debug("[FULLCONTACT.BlackList:SaveImportation]-------> RELEASE Cassandra CONNECTION !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!")
		ksOE:Close()
	end

	return success, importFormatId, err
end

--[[
	Funcion que retorna true, pero la dejo aca por si tenemos que realizar alguna validacion en algun momento.
]]--
public.BlackList.CanRemoveBlackListAddress = function(blackListAddress, vcc)
	return true
end

---
--
public.BlackList.CanRemoveImport = function(blackListImportId, ksOE)
	local isBlackList = true
	local success, blacklistImportInfo = public.Dialer.CanRemoveOrDepurateImport(blackListImportId, isBlackList, ksOE)
	if not success then
		-- error
		local err = blacklistImportInfo
		logger.warn("[FULLCONTACT:BlackList:CanRemoveImport] 1- Unable to remove blacklist Importation '%s'. Err: %s", blackListImportId, json.encode(err))
		return false, err
	end
	logger.debug("[FULLCONTACT:BlackList:CanRemoveImport] blackListImportId: '%s'. blacklistImportInfo.Campaigns: %s", blackListImportId, json.encode(blacklistImportInfo.Campaigns))
	if next(blacklistImportInfo.Campaigns or {}) then
		-- error
		local err = {err = "BLACKLIST ASSIGNED IN SOME CAMPAIGNS", importId = blackListImportId, type = -5, campaigns = blacklistImportInfo.Campaigns}
		logger.warn("[FULLCONTACT:BlackList:CanRemoveImport] 2- Unable to remove blacklist Importation '%s'. The blacklist is in use in some campaigns. Err: %s", blackListImportId, json.encode(err))
		return false, err
	end
	-- ok
	logger.info("[FULLCONTACT.BlackList:CanRemoveImport] Blacklist '%s' can be removed.", blackListImportId)
	return true, blacklistImportInfo
end

---
--
public.BlackList.RemoveImport = function(blackListImportId, userId, pKsOE)
	assert(type(blackListImportId) == "string")

	blackListImportId = blackListImportId:lower()
	userId = userId or "ANONYMUS"

	logger.info("[FULLCONTACT.BlackList:RemoveImport] Wait moment please ! ... Removing BlackList IMPORTATION '%s'. IMPORTATION_TYPE '%s'.", blackListImportId, IMPORTATION_TYPE)

	local ksOE = pKsOE or m_datasource.cassandra:CassandraConnection("OutboundEngine")
	local blacklistImportInfo
	local function CanRemoveImportation ()
		-- valido si existe antes de remover
		local success
		success, blacklistImportInfo = public.BlackList.CanRemoveImport(blackListImportId, ksOE)
		return success, blacklistImportInfo
	end

	local statusInfo = {
		status = "DELETING",
		action = "START PROCESS",
		added = 0,
		updated = 0,
		errors = 0,
		deleted = 0
	}

	local function UpdateVCCStatusProcessBatch (updateOnlyRealtime)
		public.BlackList:SaveImportation(blacklistImportInfo, statusInfo, updateOnlyRealtime, nil, ksOE)
	end

	local deletedAddresses = {}

	local function RollBackAddresses (errRemoveContact)
		-- CANCELO si es que existe al menos 1 address que no se puede remover.
		statusInfo.status = "COMPLETE"
		statusInfo.action = "ENDED DELETING"
		UpdateVCCStatusProcessBatch()

		local err = errRemoveContact or {code = "UNABLE TO REMOVE IMPORTATION", type = "-3", reason = "there are addresses that cannot be removed"}
		logger.warn("[FULLCONTACT.BlackList:RemoveImport]---> Unable to remove importation: '%s' vcc: '%s'. %s.", blackListImportId, blacklistImportInfo.VCC, json.encode(err))

		local addressesToRollback = {}
		for _, deletedAddress in pairs(deletedAddresses or {}) do
			local rollBack = {
				action  = "CREATE",
				address = deletedAddress
			}
			--logger.debug("[FULLCONTACT.BlackList:RemoveImport.RollBackAddresses] *****************---> rollBack: '%s'.", helpers.JSON_Encode(rollBack))
			table.insert(addressesToRollback, rollBack)
		end
		local success, result = ES.BlackList:UpdateAddressesBatch(blacklistImportInfo.VCC, addressesToRollback)
		return false, err
	end

	local function RemoveAddressesImport ()
		require("socket").sleep(1) -- hay que esperar un tiempo determinado (1 segundo) para que el delete se haga efectivo y luego realizar la siguiente consulta.
		local success, addresses = public.BlackList.LoadPaginAddressesOfImportation(blackListImportId)
		if not success or (success and #addresses == 0) then
			logger.info("[FULLCONTACT.BlackList:RemoveImport] No more addresses in Importation '%s' to remove.", blackListImportId)
			return true, {code = "NO MORE ADDRESSES"}
		end

		-- 2 - valido si los addresses pueden ser removidos.
		addresses = addresses or {}
		local canRemovePaginAddresses = true
		local errRemoveAddress
		for _, address in ipairs(addresses) do
			if not deletedAddresses[address.Key] then
				local canRemove, err = public.BlackList.CanRemoveAddress(address, blacklistImportInfo.VCC)
				if not canRemove then
					canRemovePaginAddresses = false
					errRemoveAddress = err
					break
				end
			-- else
			-- 	logger.debug("[FULLCONTACT.BlackList:RemoveImport] --> Already deleted address '%s'.", address.Id)
			end
		end

		if not canRemovePaginAddresses then
			-- CANCELO si es que existe al menos 1 address que no se puede remover.
			return RollBackAddresses(errRemoveAddress)
		end

		-- 3 - Elimino de ES los addresses
		local updateStatusCounter = m_configValues.importation.updateStatusCounter
		local counter = 0
		local addressesToRemove = addresses

		-- BATCH (Muchos)
		--
		local success, result, detail = ES.BlackList:RemoveAddressesBatch(blacklistImportInfo.VCC, addressesToRemove)
		-- logger.debug("[FULLCONTACT:RemoveImport] 0.5- addressesToRemove '%s'.", #addressesToRemove)
		-- logger.debug("[FULLCONTACT:RemoveImport] 1- success '%s'.", success)
		-- logger.debug("[FULLCONTACT:RemoveImport] 2- result  '%s'.", json.encode(result))
		-- logger.debug("[FULLCONTACT:RemoveImport] 3- detail  '%s'.", json.encode(detail))
		if success then
			-- 2.
			--
			-- --if detail.delete and (detail.delete.deleted or 0) < #addressesToRemove then
			-- 	for i, address in ipairs(addressesToRemove) do
			-- 		deletedAddresses[address.Key] = address
			-- 	end
			-- 	return RollBackAddresses()
			-- --end
		else
			-- 3. si es error, actualiza los contadores (errors + los invalid para descargar el CSV)
			--
			return RollBackAddresses()
		end
		for i, address in ipairs(addressesToRemove) do
			counter = counter + 1
			if counter == updateStatusCounter then
				statusInfo.status = "DELETING"
				statusInfo.action = "DELETED ADDRESS: " .. tostring(i)
				UpdateVCCStatusProcessBatch()
				counter = 0
			end
			deletedAddresses[address.Key] = address
		end

		-- ok
		if #addresses < ROWS_PER_PAGE then
			-- more addresses
			--logger.info("[FULLCONTACT.BlackList:RemoveImport] -- No more addresses in Importation '%s' to remove.", blackListImportId)
			return true, {code = "NO MORE ADDRESSES"}
		end
		-- MORE ADDRESSES
		return true, {code = "MORE ADDRESSES"}
	end

	local function StartRemoveImport ()
		local success, err = CanRemoveImportation()
		if not success then
			return false, err
		end

		-- 1 - Obtengo los addresses de la importación.
		UpdateVCCStatusProcessBatch(true)
		local canRemoveImportation = false
		local currentLoadedRows = 0
		local err
		while true do
			currentLoadedRows = currentLoadedRows + ROWS_PER_PAGE
			local success, result = RemoveAddressesImport()
			if success then
				if result.code == "NO MORE ADDRESSES" then
					--logger.debug("[FULLCONTACT.BlackList:RemoveImport] 1 OK ---> NO MORE ADDRESSES. total loaded: %s", currentLoadedRows)
					canRemoveImportation = true
					break
				--elseif result.code == "MORE ADDRESSES" then
					--logger.debug("[FULLCONTACT.BlackList:RemoveImport] 2 OK ---> MORE ADDRESSES. loaded: %s", currentLoadedRows)
				--else
					--logger.debug("[FULLCONTACT.BlackList:RemoveImport] 3 OK ---> Unrecognized code: %s. loaded: %s", result.code, currentLoadedRows)
				end
			else
				--logger.debug("[FULLCONTACT.BlackList:RemoveImport] 4 FAIL ---> Unable to delete addresses, code: %s. loaded: %s", result.code, currentLoadedRows)
				canRemoveImportation = false
				err = result or {err = "UNABLE TO COMPLETE DELETE ADDRESSES OF IMPORTAION"}
				err.importId = blackListImportId
				err.type = err.type or "-4"
				break
			end
			if currentLoadedRows >= (MAXIMUM_NUMBER_OF_ROWS_IN_DIALER_IMPORT + ROWS_PER_PAGE) then
				canRemoveImportation = true
				--logger.debug("[FULLCONTACT.BlackList:RemoveImport] 5 FAIL ---> Unable to delete addresses: OVERFLOW. loaded: %s", currentLoadedRows)
				break
			end
		end

--local canRemoveImportation = true
		if not canRemoveImportation then
			-- ERROR
			return false, err
		end
		-- OK
		--
		-- 4 - Actualizo el estado en REDIS.
		statusInfo.status = "COMPLETE"
		statusInfo.action = "ENDED DELETING"
		UpdateVCCStatusProcessBatch(true)
		-- 5 - Elimino de CASSANDRA la Importación.
		ksOE:RemoveColumn(blackListImportId, "BlackListImport")
		-- 6 - Elimino de CASSANDRA las Addresses fallidas de la Importación.
		ksOE:RemoveColumn(blackListImportId, "BlackListFailedAddresses")
		-- 7 - Elimino la relacion BlacklistId que se encuentra en la CF ContactImport (importacion de contactos)
		for contactImportId in pairs(blacklistImportInfo.AssociatedContactsImports or {}) do
			local success, contactsImportInfo = public.Dialer.GetImportation(contactImportId, nil, ksOE)
			if success then
				if contactsImportInfo.Blacklists then
					contactsImportInfo.Blacklists[blackListImportId] = nil
					public.Dialer.SaveContactImportInDB(contactImportId, contactsImportInfo, ksOE)
				end
			end
		end
		return true
	end

	local success, resp = StartRemoveImport()
	if not pKsOE then
		ksOE:Close()
	end

	return success, resp
end

---
--
public.BlackList.GetImportFailedAddresses = function(blackListImportId, detailId)
	assert(type(blackListImportId) == "string")
	assert(type(detailId) == "string")

	blackListImportId = blackListImportId:lower()
	detailId = detailId:lower()

	--logger.info("[FULLCONTACT.BlackList:GetImportFailedAddresses] Wait moment please ... Retrieving BlackList Import Failed addresses. Id '%s' detailId: '%s'.", blackListImportId, detailId)

	local importDetailsFailedAddresses = helpers.GetAllColumns {
		keyspace = "OutboundEngine",
		columnFamily = "BlackListFailedAddresses",
		superColumns = true,
		key = blackListImportId
	}

	--logger.debug("[FULLCONTACT.BlackList:GetImportFailedAddresses] Retrieved BlackList Import Failed Details from Cassandra '%s'.", helpers.JSON_Encode(importDetailsFailedContacts or {}))

	if helpers.TableIsEmpty(importDetailsFailedAddresses) then
		--logger.info("[FULLCONTACT:BlackList:GetImportFailedAddresses] Importation '%s' no has failed addresses.", blackListImportId)
		return false
	end

	local failedAddresses = importDetailsFailedAddresses[detailId]
	if not failedAddresses then
		--logger.info("[FULLCONTACT:BlackList:GetImportFailedAddresses] Importation '%s' no hast failed addresses detailId: %s.", blackListImportId, detailId)
		return false
	end

	--logger.info("[FULLCONTACT.BlackList:GetImportFailedAddresses] Is Found the BlackList Import Failed addresses. Id '%s' detailId: '%s'.", blackListImportId, detailId)
	return true, failedAddresses
end

public.BlackList.SaveAddressesBatch = function (self, vcc, addresses, options)
	--logger.debug("[FULLCONTACT.BlackList:SaveAddressesBatch] -> Saving '%s' addresses BATCH in vcc.", #addresses, vcc)
	--logger.debug("[FULLCONTACT.BlackList:SaveAddressesBatch] - Saving addresses BATCH '%s'.", helpers.JSON_Encode(addresses))
	options = options or {}
	local success, result, detail = ES.BlackList:ValidateDupAndStoreAddressesBatch(vcc, addresses, options)
	if not success then
		-- error
		local err = result
		return false, err, detail
	end
	-- ok
	return true, result, detail
end

---
--
public.BlackList.LoadPaginAddressesOfImportation = function(importId)
	assert(type(importId) == "string")
	importId = importId:lower()
	local rowsPerPage = ROWS_PER_PAGE

	--logger.info("[FULLCONTACT.BlackList:LoadPaginAddressesOfImportation] Wait moment please ... Loading %d BlackList Dialer of Importation '%s'.", rowsPerPage, importId)

	local success, importationInfo = public.BlackList.GetImportation(importId)
	if not success then
		logger.error("[FULLCONTACT:BlackList:LoadPaginAddressesOfImportation] Unable to retrieve Importation '%s'. Does not exist", importId)
		return false, {err = "IMPORTATION DOES NOT EXIST", importId = importId, type = "-1"}
	end

	local importIdWithoutVCC = helpers.SeparateVCC(importId)
	local function ExecuteQuery (startRow)
		local query = {
			filter = {
				VCC = {
					exact  = true,
					values = {importationInfo.VCC}
				},
				ImportId = {
					exact  = true,
					values = {importIdWithoutVCC}
				}
			},
			--vcc         = importationInfo.VCC,
			startRow    = startRow,
			rowsPerPage = rowsPerPage
		}
		local success, response = ES.BlackList:SearchAddresses(query)
		--logger.debug("[FULLCONTACT:BlackList:LoadPaginAddressesOfImportation] Loaded dialer BlackList of importation '%s' vcc: '%s'. success: %s, response: %s", importId, importationInfo.VCC, success, helpers.JSON_Encode(response or {}))
		return success, response
	end

	local from = 0
	local success, response = ExecuteQuery(from)
	if not success then
		return false, response
	end
	local addresses = {}
	local responseAddresses = response.addresses or {}
	for _, address in ipairs(responseAddresses) do
		table.insert(addresses, address)
	end

	--logger.debug("[FULLCONTACT:BlackList:LoadPaginAddressesOfImportation] --> Loading %d BlackList dialer of importation '%s' vcc: '%s'.", #addresses, importId, importationInfo.VCC)

	return true, addresses
end

--[[
	Funcion que retorna true, pero la dejo aca por si tenemos que realizar alguna validacion en algun momento.
]]--
public.BlackList.CanRemoveAddress = function(address, vcc)
	assert(type(address) == "table")
	return true
end

public.BlackList.SearchAddresses = function (self, params)
	local vcc =  params.vcc or params.VCC
	--logger.info("[FULLCONTAC.BlackList:SearchAddresses] - Searching BlackList Addresses in VCC '%s' using ELASTIC SEARCH: filter '%s', start '%s', rows per page '%s', sorted: '%s'.", vcc, helpers.JSON_Encode(params.filter), params.startRow, params.rowsPerPage, (params.sort and helpers.JSON_Encode(params.sort) or "NONE"))
	local query = {
		filter = params.filter,
		startRow = params.startRow,
		rowsPerPage = params.rowsPerPage,
		VCC = vcc
	}
	local success, response = ES.BlackList:SearchAddresses(query)
	if not success then
		local err = response
		return false, err
	end
	return true, response
end

public.BlackList.GetAddress = function (self, addressESKey)
	assert(type(addressESKey) == "string")
	--logger.info("[FULLCONTACT.BlackList:GetAddress] - Get address - addressESKey: '%s'.", addressESKey)
	return ES.BlackList:GetAddress(addressESKey)
end

--[[
	La funcion queda por si es necesario realizar alguna implementacion
]]--
public.BlackList.CanUpdateRelatedBlacklistsWithCampiagnsAccounts = function()
	return true
end

public.BlackList.MakeBlacklistId = function(sourceId, isGlobal, vcc)
	assert(type(sourceId) == "string")
	local targetId
	if isGlobal then
		targetId = helpers.AddVCC(sourceId, "system")
	else
		assert(type(vcc) == "string")
		targetId = helpers.AddVCC(sourceId, vcc)
	end
	logger.debug("[FULLCONTACT.BlackList:MakeBlacklistId] --> sourceId: '%s', targetId: '%s'.", sourceId, targetId)
	return targetId
end

--[[
]]--
public.BlackList.ParseScopeId = function(self, scopeId)
	assert(type(scopeId) == "string")
	local separatorPos = string.find(scopeId, ":")
	if not separatorPos then
		--LogDebug("[FULLCONTACT.BlackList:ParseScopeId] *--> scopeId: %s', vcc: %s', id: %s'.", scopeId, scopeId, scopeId)
		return scopeId, scopeId
	end
	local vcc = string.sub(scopeId, 1, separatorPos - 1)
	local id = string.sub(scopeId, separatorPos + 1)
	--LogDebug("[FULLCONTACT.BlackList:ParseScopeId] --> scopeId: %s', vcc: %s', id: %s'.", scopeId, vcc, id)
	return vcc, id
end

--[[
	Funcion que actualiza la relacion de blacklist con las campanas->cuentas.
]]--
public.BlackList.AddRelatedCampaignsAccounts = function(self, vcc, campaignId, accountId, oldBlacklists, newBlacklists, userId, pKsOE)
	userId = userId or "ANONYMUS"

	LogInfo("[FULLCONTACT.BlackList:AddRelatedCampaignsAccounts] Wait moment please ... Updating Relation Blacklists with Campaigns Accounts .. UserId: %s'.", userId)
	--LogDebug("[FULLCONTACT.BlackList:AddRelatedCampaignsAccounts] 1 - oldBlacklists: %s'.", json.encode(oldBlacklists or {}))
	--LogDebug("[FULLCONTACT.BlackList:AddRelatedCampaignsAccounts] 2 - newBlacklists: %s'.", json.encode(newBlacklists or {}))

	-- 1. valido si es posible realizar las relaciones.
	local success, resp = public.BlackList.CanUpdateRelatedBlacklistsWithCampiagnsAccounts()
	if not success then
		local err = resp
		return false, err
	end

	-- 2. en la lista "unassignedBlacklists" nos quedamos con las currentBlacklists que se han quitado de la campana->cuenta
	local unassignedBlacklists = {}
	for scopeId, info in pairs(oldBlacklists) do
		local _, id = self:ParseScopeId(scopeId)
		local assignedInfo = newBlacklists[id]
		if not assignedInfo then
			unassignedBlacklists[id] = info
		end
	end
	--logger.debug("[FULLCONTACT.BlackList:AddRelatedCampaignsAccounts] 1 - unassignedBlacklists: %s'.", json.encode(unassignedBlacklists))

	-- 3. recorro la lista "unassignedBlacklists" y quito las campañas y cuentas que no usan mas la blacklist, luego se actualiza  la cf BlacklistImmport
	local ksOE = pKsOE or m_datasource.cassandra:CassandraConnection("OutboundEngine")
	for scopeId, info in pairs(unassignedBlacklists) do
		local _, id = self:ParseScopeId(scopeId)
		local blacklistId = public.BlackList.MakeBlacklistId(id, info.isGlobal, vcc)
		local success, blackListImport = public.BlackList.GetImportation(blacklistId, nil, ksOE)
		if success then
			if blackListImport.Campaigns and blackListImport.Campaigns[campaignId] then
				blackListImport.Campaigns[campaignId][accountId] = nil
				if not next(blackListImport.Campaigns[campaignId]) then
					blackListImport.Campaigns[campaignId] = nil
				end
				public.BlackList.Save(blacklistId, blackListImport, ksOE)
			end
		else
			logger.warn("[FULLCONTACT.BlackList:AddRelatedCampaignsAccounts] Blacklist '%s' does not exist. Unable to unassign campaign: '%s', account: '%s'.", id, campaignId, accountId)
		end
	end

	-------------
	-- 4. en la lista "newAssignedBlacklists" nos quedamos con las nuevas blacklists que se han agregado en la campana->cuenta
	local newAssignedBlacklists = {}
	for scopeId, info in pairs(newBlacklists) do
		local _, id = self:ParseScopeId(scopeId)
		local assignedInfo = oldBlacklists[id]
		if not assignedInfo then
			newAssignedBlacklists[id] = info
		end
	end
	--logger.debug("[FULLCONTACT.BlackList:AddRelatedCampaignsAccounts] 2 - newAssignedBlacklists: %s'.", json.encode(newAssignedBlacklists))

	-- 5. recorro la lista "newAssignedBlacklists" y agrego las campañas y cuentas que se han incluido en la blacklist, luego se actualiza  la cf BlacklistImmport
	for scopeId, info in pairs(newAssignedBlacklists) do
		local _, id = self:ParseScopeId(scopeId)
		local blacklistId = public.BlackList.MakeBlacklistId(id, info.isGlobal, vcc)
		local success, blackListImport = public.BlackList.GetImportation(blacklistId, nil, ksOE)
		if success then
			blackListImport.Campaigns = blackListImport.Campaigns or {}
			blackListImport.Campaigns[campaignId] = blackListImport.Campaigns[campaignId] or {}
			blackListImport.Campaigns[campaignId][accountId] = blackListImport.Campaigns[campaignId][accountId] or {}
			public.BlackList.Save(blacklistId, blackListImport, ksOE)
		else
			logger.warn("[FULLCONTACT.BlackList:AddRelatedCampaignsAccounts] Blacklist '%s' does not exist. Unable to assign campaign: '%s', account: '%s'.", id, campaignId, accountId)
		end
	end

	-- 6. codigo necesario para actualizar en builds previos a la primera instalacion
	if next(newBlacklists) and not next(newAssignedBlacklists) then
		logger.debug("[FULLCONTACT.BlackList:AddRelatedCampaignsAccounts] 3 - Updating for previous builds.")
		for scopeId, info in pairs(newBlacklists) do
			local _, id = self:ParseScopeId(scopeId)
			local blacklistId = public.BlackList.MakeBlacklistId(id, info.isGlobal, vcc)
			local success, blackListImport = public.BlackList.GetImportation(blacklistId, nil, ksOE)
			if success and not blackListImport.Campaigns then
				blackListImport.Campaigns = {}
				blackListImport.Campaigns[campaignId] = blackListImport.Campaigns[campaignId] or {}
				blackListImport.Campaigns[campaignId][accountId] = blackListImport.Campaigns[campaignId][accountId] or {}
				public.BlackList.Save(blacklistId, blackListImport, ksOE)
			else
				logger.warn("[FULLCONTACT.BlackList:AddRelatedCampaignsAccounts] Blacklist '%s' does not exist. Unable to assign campaign: '%s', account: '%s' for previous builds.", id, campaignId, accountId)
			end
		end
	end

	if not pKsOE then
		ksOE:Close()
	end

	return true
end

--[[
	Funcion que elimina la relacion de blacklist con las campanas->cuentas.
]]--
public.BlackList.RemoveRelatedCampaignsAccounts = function(self, vcc, campaignId, accountId, blacklists, userId, pKsOE)
	userId = userId or "ANONYMUS"

	logger.info("[FULLCONTACT.BlackList:RemoveRelatedCampaignsAccounts] Removing Relation Blacklists with Campaigns Accounts. campaignId: %s, accountId: %s, vcc: %s,  userId: %s'.", campaignId, accountId, vcc, userId)
	--LogDebug("[FULLCONTACT.BlackList:RemoveRelatedCampaignsAccounts] 1 - blacklists: %s'.", json.encode(blacklists or {}))

	-- 1. valido si es posible realizar las relaciones.
	local success, resp = public.BlackList.CanUpdateRelatedBlacklistsWithCampiagnsAccounts()
	if not success then
		local err = resp
		return false, err
	end

	local ksOE = pKsOE or m_datasource.cassandra:CassandraConnection("OutboundEngine")
	for scopeId, info in pairs(blacklists) do
		local _, id = self:ParseScopeId(scopeId)
		local blacklistId = public.BlackList.MakeBlacklistId(id, info.isGlobal, vcc)
		local success, blackListImport = public.BlackList.GetImportation(blacklistId, nil, ksOE)
		if success then
			if blackListImport.Campaigns and blackListImport.Campaigns[campaignId] then
				local hasRemoveCampaignRelation = false
				if accountId then
					blackListImport.Campaigns[campaignId][accountId] = nil
					if not next(blackListImport.Campaigns[campaignId]) then
						hasRemoveCampaignRelation = true
					end
				else
					hasRemoveCampaignRelation = true
				end
				if hasRemoveCampaignRelation then
					blackListImport.Campaigns[campaignId] = nil
				end
				public.BlackList.Save(blacklistId, blackListImport, ksOE)
			end
		else
			logger.warn("[FULLCONTACT.BlackList:AddRelatedCampaignsAccounts] Blacklist '%s' does not exist. Unable to unassign campaign: '%s', account: '%s'.", id, campaignId, accountId)
		end
	end
	if not pKsOE then
		ksOE:Close()
	end

	return true
end

-------------------------
-- Campaigns
--------------------------
public.Campaigns = {}

--[[
]]--
public.Campaigns.GetVCCCampaigns = function(self, vcc, ksAgent)
	return helpers.GetAllColumns {
		keyspace = ksAgent or "Agent",
		columnFamily = "VCCCampaigns",
		key = vcc
	} or {}
end

--[[
]]--
public.Campaigns.GetCampaignAccounts = function(self, campaignId, vcc, ksAgent)
	local fullCampaignId = helpers.AddVCC(campaignId, vcc)
	local params = {
		keyspace = keyspace or "Agent",
		superColumns = true,
		columnFamily = "CampaignAccounts",
		key = fullCampaignId
	}
	return helpers.GetAllColumns(params) or {}, fullCampaignId
end

-------------------------
return public
